clanker-saddle/src/Services/SaddlePollState.vala

using Invercargill.DataStructures;
using Inversion;

namespace ClankerSaddle.Services {

    /**
     * Process-wide saddle poll bookkeeping: the last daemon-side output
     * entry id stored per session (so redelivered results pushes never
     * duplicate `session_commands` rows), the block answers and human
     * comments already handed to a saddle (each delivered exactly once;
     * a restart re-delivers and the daemon deduplicates), plus the
     * plan-10 ask/wake liveness bookkeeping: the outcome of a saddle's
     * last `POST /api/saddle/next-task` ask (a saddle counts as SLEEPING
     * when its last ask returned 204) and the pending wake each sleeping
     * saddle was sent (the lost-wake guard's timer), and the phase-2
     * push-liveness clock: the last chunks/results arrival per session
     * (the staleness probe's timer — one exceptional probe per silent
     * episode, reset by the next push).
     *
     * A singleton by necessity: {@link SaddleService} is request-scoped,
     * so this object is where the cross-request cursors live. Ids are
     * stringified: the Invercargill collections cannot key on value types.
     */
    public class SaddlePollState : Object {

        private Dictionary<string, string> probe_state = new Dictionary<string, string>();
        private Dictionary<string, string> probe_uptime = new Dictionary<string, string>();
        private Dictionary<string, string> probe_activity = new Dictionary<string, string>();
        private Dictionary<string, string> probe_contact = new Dictionary<string, string>();
        private Dictionary<string, string> last_stored_output_id = new Dictionary<string, string>();
        private HashSet<string> delivered_answers = new HashSet<string>();
        private HashSet<string> delivered_comments = new HashSet<string>();
        private HashSet<string> resumed_notified = new HashSet<string>();
        private Dictionary<string, string> ask_outcome = new Dictionary<string, string>();
        private HashSet<string> ask_seen = new HashSet<string>();
        private Dictionary<string, string> wake_at_ms = new Dictionary<string, string>();
        private Dictionary<string, string> wake_task = new Dictionary<string, string>();
        private Dictionary<string, string> push_at_ms = new Dictionary<string, string>();
        private HashSet<string> push_seen = new HashSet<string>();
        private HashSet<string> stale_probed = new HashSet<string>();
        private Mutex state_lock = Mutex();

        /** The last output entry id stored for `session_id` (0 when none). */
        public int64 since_for(int64 session_id) {
            state_lock.lock();
            string stored = "";
            bool found = last_stored_output_id.try_get(session_id.to_string(), out stored);
            state_lock.unlock();
            return found ? int64.parse(stored) : 0;
        }

        /** Records the new high-water output entry id for `session_id`. */
        public void store_since(int64 session_id, int64 entry_id) {
            state_lock.lock();
            last_stored_output_id.set(session_id.to_string(), entry_id.to_string());
            state_lock.unlock();
        }

        /**
         * Records a daemon /status observation for the saddles-page
         * health panel: reported state, uptime, last-activity epoch and
         * the web-side contact time (monotonic ms).
         */
        public void record_probe(int64 saddle_id, string state, int64 uptime_seconds,
                int64 last_activity_epoch) {
            state_lock.lock();
            string key = saddle_id.to_string();
            probe_state.set(key, state);
            probe_uptime.set(key, uptime_seconds.to_string());
            probe_activity.set(key, last_activity_epoch.to_string());
            probe_contact.set(key, (get_monotonic_time() / 1000).to_string());
            state_lock.unlock();
        }

        /** The last probe snapshot for `saddle_id` ("" when never probed). */
        public void health_for(int64 saddle_id, out string state, out int64 uptime,
                out int64 activity, out int64 contact_ms) {
            state_lock.lock();
            string key = saddle_id.to_string();
            string stored = "";
            state = probe_state.try_get(key, out stored) ? stored : "";
            uptime = int64.parse(probe_uptime.get_or_default(key) ?? "0");
            activity = int64.parse(probe_activity.get_or_default(key) ?? "0");
            contact_ms = int64.parse(probe_contact.get_or_default(key) ?? "0");
            state_lock.unlock();
        }

        /**
         * Marks `session_id`'s daemon-side resume as notified; returns
         * false when the progress entry already went out.
         */
        public bool mark_resumed(int64 session_id) {
            state_lock.lock();
            bool already = resumed_notified.has(session_id.to_string());
            if (!already) {
                resumed_notified.add(session_id.to_string());
            }
            state_lock.unlock();
            return !already;
        }

        /**
         * Marks `block_id`'s answer as delivered; returns false when it
         * had already been handed over.
         */
        public bool mark_delivered(int64 block_id) {
            state_lock.lock();
            bool already = delivered_answers.has(block_id.to_string());
            if (!already) {
                delivered_answers.add(block_id.to_string());
            }
            state_lock.unlock();
            return !already;
        }

        /** True when `block_id`'s answer was already handed over. */
        public bool is_delivered(int64 block_id) {
            state_lock.lock();
            bool already = delivered_answers.has(block_id.to_string());
            state_lock.unlock();
            return already;
        }

        /**
         * Marks `comment_id` as delivered to its saddle; returns false
         * when it had already been handed over.
         */
        public bool mark_comment_delivered(int64 comment_id) {
            state_lock.lock();
            bool already = delivered_comments.has(comment_id.to_string());
            if (!already) {
                delivered_comments.add(comment_id.to_string());
            }
            state_lock.unlock();
            return !already;
        }

        /** True when `comment_id` was already handed over. */
        public bool is_comment_delivered(int64 comment_id) {
            state_lock.lock();
            bool already = delivered_comments.has(comment_id.to_string());
            state_lock.unlock();
            return already;
        }

        /**
         * Records an ask arriving from `saddle_id` (plan 10): the
         * pending wake is answered — the lost-wake guard must not fire
         * for it — and the saddle is awake until the ask resolves. The
         * first ask also marks the daemon ask-capable (post-flip): its
         * parked-task resumes flow through wake + next-task from then
         * on, never through /start pushes.
         */
        public void mark_ask(int64 saddle_id) {
            state_lock.lock();
            string key = saddle_id.to_string();
            ask_outcome.set(key, "awake");
            ask_seen.add(key);
            wake_at_ms.remove(key);
            wake_task.remove(key);
            state_lock.unlock();
        }

        /**
         * Records the outcome of `saddle_id`'s ask: `204` puts it to
         * sleep (a later wake targets it), anything else leaves it
         * awake.
         */
        public void mark_ask_result(int64 saddle_id, bool assigned) {
            state_lock.lock();
            ask_outcome.set(saddle_id.to_string(), assigned ? "assigned" : "204");
            state_lock.unlock();
        }

        /**
         * True when `saddle_id`'s last ask returned 204 — the sleeping
         * state a wake targets.
         */
        public bool is_sleeping(int64 saddle_id) {
            state_lock.lock();
            string stored = "";
            bool sleeping = ask_outcome.try_get(saddle_id.to_string(), out stored)
                && stored == "204";
            state_lock.unlock();
            return sleeping;
        }

        /**
         * True once `saddle_id` has ever asked (post-flip daemon): it
         * asks whenever idle-awake, so its work — fresh or resumed —
         * always flows through the allocation.
         */
        public bool has_ever_asked(int64 saddle_id) {
            state_lock.lock();
            bool seen = ask_seen.has(saddle_id.to_string());
            state_lock.unlock();
            return seen;
        }

        /** Forgets the ask state (the saddle went offline / was removed). */
        public void clear_ask(int64 saddle_id) {
            state_lock.lock();
            string key = saddle_id.to_string();
            ask_outcome.remove(key);
            wake_at_ms.remove(key);
            wake_task.remove(key);
            state_lock.unlock();
        }

        /**
         * Records a wake sent to `saddle_id` for `task_id` — starts the
         * lost-wake guard's timer.
         */
        public void note_wake(int64 saddle_id, int64 task_id) {
            state_lock.lock();
            string key = saddle_id.to_string();
            wake_at_ms.set(key, (get_monotonic_time() / 1000).to_string());
            wake_task.set(key, task_id.to_string());
            state_lock.unlock();
        }

        /**
         * True when `saddle_id` has a pending wake at least
         * `guard_ms` old — an ask should have arrived by now; the
         * guard declares the saddle offline and frees the work for
         * the next capable one.
         */
        public bool wake_overdue(int64 saddle_id, uint guard_ms) {
            state_lock.lock();
            string stored = "";
            bool overdue = wake_at_ms.try_get(saddle_id.to_string(), out stored)
                && (get_monotonic_time() / 1000 - int64.parse(stored)) >= guard_ms;
            state_lock.unlock();
            return overdue;
        }

        /**
         * True when `saddle_id` carries a pending wake younger than
         * `guard_ms` — the wake pass skips it instead of re-waking on
         * every tick while the guard window runs.
         */
        public bool wake_pending(int64 saddle_id, uint guard_ms) {
            state_lock.lock();
            string stored = "";
            bool pending = wake_at_ms.try_get(saddle_id.to_string(), out stored)
                && (get_monotonic_time() / 1000 - int64.parse(stored)) < guard_ms;
            state_lock.unlock();
            return pending;
        }

        /** The task id `saddle_id` was last woken for (0 when none). */
        public int64 wake_task_of(int64 saddle_id) {
            state_lock.lock();
            string stored = "";
            bool found = wake_task.try_get(saddle_id.to_string(), out stored);
            state_lock.unlock();
            return found ? int64.parse(stored) : 0;
        }

        /**
         * Records a chunks/results push arriving for `session_id`
         * (plan 10 phase 2): the running daemon is alive, and any
         * running staleness episode ends — a later silent episode
         * earns its own exceptional probe.
         */
        public void note_push(int64 session_id) {
            state_lock.lock();
            string key = session_id.to_string();
            push_at_ms.set(key, (get_monotonic_time() / 1000).to_string());
            push_seen.add(key);
            stale_probed.remove(key);
            state_lock.unlock();
        }

        /**
         * Atomically decides the one exceptional staleness probe for
         * `session_id`: true exactly when pushes have been seen, the
         * silence has reached `stale_ms` and no probe fired for this
         * episode yet (the flag flips here, so a slow probe is never
         * double-fired by the next tick). A session with no recorded
         * push (a web restart wiped the clock) is seeded now and
         * starts its window instead of firing immediately — a fresh
         * baseline must not masquerade as 90 s of silence.
         */
        public bool take_stale_probe(int64 session_id, uint stale_ms) {
            state_lock.lock();
            string key = session_id.to_string();
            int64 now_ms = get_monotonic_time() / 1000;
            if (!push_seen.has(key)) {
                push_seen.add(key);
                push_at_ms.set(key, now_ms.to_string());
                state_lock.unlock();
                return false;
            }
            if (stale_probed.has(key)) {
                state_lock.unlock();
                return false;
            }
            string stored = "";
            bool known = push_at_ms.try_get(key, out stored);
            bool due = known && (now_ms - int64.parse(stored)) >= stale_ms;
            if (due) {
                stale_probed.add(key);
            }
            state_lock.unlock();
            return due;
        }
    }
}