Bound every action-log read path - #298
Conversation
7a6dcf3 to
eef947d
Compare
03c86bd to
eef947d
Compare
| * the log in bounded internal pages; ordering relative to live updates is simply the stream | ||
| * order), then ready() fires. Resolved history is fetched separately via listActions(). | ||
| * | ||
| * `startAfter` is deprecated: its presence switches the replay from pending-only to every |
There was a problem hiding this comment.
Why this design change to subscriptions?
subscribeToActions() was not the problem before, since it is only used to subscribe to new actions as they happen. (The startAfter parameter is meant for re-establishing a subscription after disconnecting, without missing anything. It would normally be set to the date of the last action seen.)
With the change, it seems it is no longer possible to subscribe to observations, since they are never "pending".
There was a problem hiding this comment.
Hmmm that may have been how it was designed, but that does not match how it was implemented. On current main, useActions always passes new Date(0) on every store open. There's even a comment that leads me to think this was intentional
"startAfter: epoch 0 asks the backend to replay full history through the subscriber, avoiding a stale-state race against a separate listActions()."
and even with the intended approach it is still a O(entire log) since records aren't indexed by time and overseer does for (let record of actions.list())
"It is no longer possible to subscribe to observations, since they are never 'pending'"
It is :)
Give it a whirl on the preview branch and you'll see this does work
Only the replay becomes pending only - no changes to live subscription
Fair point buried in there though: the subscription by itself no longer catches you up on stuff that resolved while you were disconnected. I moved that job moved to listActions paging (plus the reconnect re-fetch for action cards). So the real question may be whether "subscription = live + pending, history = paged reads" is the split we want. But I think that's pretty defensible.
listActions was essentially dead code on main - it had zero callers.
There was a problem hiding this comment.
Yikes, OK, so the UI code was using a subscription-since-time-zero to backfill instead of using listActions()? That definitely wasn't the intended design. The idea was supposed to be that you use listActions() to backfill and then subscribeToActions() to receive new actions real-time.
Only the replay becomes pending only - no changes to live subscription
So subscribe() subscribes to all future actions (of any type) and also surfaces historical pending actions?
This seems like a confusing interface. Why combine these? Why not call listActions() with a filter requesting pending actions specifically?
Why are pending actions treated specially, anyway? What if the user has a million old pending actions that they intentionally never accepted or rejected? Maybe they are intentionally living in a simulation (there are valid use cases).
I think the right design here is:
subscribeToActions()stays as it is. Perhaps, we modify it to limit how far in the paststartAfteris allowed to be -- since this was only intended for resuming subscriptions. But I think for short disconnects it's really better to be able to resume the subscription rather than require a separate RPC to check for missed events.- Use
listActions()to query history, including of pending actions.
There was a problem hiding this comment.
Fair points on all counts. We're basically there now (after another round of changes):
(1) subscribe = deltas only, (2) listActions() handles all history queries including pending
subscribeToActions()now delivers live deltas only; nothing is replayed.- Pending state comes from
listActions({filter: "pending"}), same paging/cursor as the rest of history. Clients initiate the subscribe first, then pages - Backing it is a sparse pending-by-gatekeeper index, which the auto-approval drain now reads too.
On the million-pending workspace: better, but I won't claim it's solved. The wire is now bounded (pull-paged, client-clocked), but each pending page still materializes the full pending set server-side, since the index can't range-read within a group yet (no time index)
The one spot I went a different way is startAfter resume. There's a gotcha: hook enable/disable/delete mutate the record with no timestamp field at all (appliedAt only exists on type: "action"), so a resumed subscription would miss hook toggles. Resolutions are fine, they stamp appliedAt. So rather than bound it, I kept startAfter purely as a compat shim for already-deployed clients (they still get the old full replay), with a TODO to delete once they cycle out.
If we want cheap short-disconnect resume later, it may be worth considering a monotonic revision counter rather than wall-clock because of workerd's frozen clock or otherwise port the getChatTimestamp hack into something more general purpose
Opening a large workspace replayed the whole action history before the UI was usable, overloading the Durable Object. Replace the unbounded reads with two paged RPCs: - scanPendingActions(): bounded id-cursor scan for pending records, with a fixed exclusive throughId captured on the first page (subscribe-then-scan contract, mirroring getChatHistory). - listActions() (repurposed; it had zero callers): resolved history, newest-first by id, with a raw-records scan cap so filtered pages stay bounded. UseOverseerInterface answers both with inert empty terminal pages, matching the speculative-call pattern of subscribeToActions.
#drainOnce materialized the entire actions collection per drain. Page it instead: capture nextActionId as the scan bound, list one materialized page at a time (preserving the iterator-invalidation safety the old snapshot provided), and yield between pages so a long scan doesn't starve client RPCs. Actions created past the bound are folded in by drain()'s existing rerun flag via the creation path's own drain call. Gate/failure semantics are unchanged: a manual gate or a failed apply still halts the whole drain.
The history tab rendered the full replayed action log. Load it on demand
instead through a new useActionHistory hook: nothing is fetched until the
tab opens, pages continue via the server cursor ("Load older"), and live
resolutions from the shared subscription merge into the loaded id window.
Day groups now follow creation order (labels may repeat), and the counter
reports entries loaded rather than a total the client no longer knows.
The shared useActions store is untouched here; the replay itself is removed
in the next commit.
This is the commit that stops the full action-log replay on workspace open.
The shared useActions store now subscribes without startAfter (live entries
only) and pages a bounded scanPendingActions() loop for records that were
already pending. The subscription is initiated first — e-ordered calls on
one stub register the DB subscriber before the scan reads anything — and a
per-generation liveSeenIds set drops any scanned record already delivered
live (live wins; a scanned page can be stale by the time it arrives).
The store's shape changes accordingly: {status, pendingById, liveById}
instead of the full actionsById map. Consumers adapt: GadgetEditor derives
the hook signature from live entries (listHooks remains the authoritative
initial source) and pending counts from pendingById; the Activity review
tab drops its whole-store loading gate in favor of checking/error states;
ActivityNotifications says it's still checking when empty mid-scan.
useActionEntries replays only live-received records, which is sufficient
for chat: fetched messages arrive server-hydrated, and everything that
changed since arrived live.
Known quirk (accepted): openActivity routes to 'history' while the scan is
still checking and nothing pending has been found yet.
One real-DO call of scanPendingActions and listActions each, proving the @validateRpc wiring accepts the new option shapes. Placed after the reset tests so abortAllDurableObjects() doesn't tear down this session's DOs mid-flight (which leaks a canceled-context rejection).
subscribeToActions now replays currently-pending records itself: after registering the DB subscriber it sweeps the log in bounded pages with scheduler.wait(0) between them (the drainer's pattern), then fires ready(). Replay and live updates share one ordered stream, so the scanPendingActions cursor/throughId protocol, the client scan loop, and its live-wins dedup all go away; ready() = the subscribe RPC resolving. The legacy startAfter full-log replay is removed outright (that path IS the overload bug); the parameter stays in the signature, ignored, for stale in-flight clients. listActions loses its client-supplied limit — page size is the server constant.
subscribeToChat hydrated actionLog only in the storage add() hook, so the update() leg and the reconnect catch-up scan delivered action messages with actionLog undefined, which the client renders as a blank card. Hydrate in deliverMessage so all three legs share it.
The pending-only replay never mentions an action that resolved while the client was away, so a cached card could stay 'pending' forever and keep the composer blocked. On each new overseer stub, sweep the cached action-message index and re-fetch cards that are blank or still pending via getChatMessage, guarding against regressing a card already resolved by a faster channel.
Pre-deploy clients pass startAfter and build their whole history view from replay; the pending-only replay left them rendering an empty activity log with no way to nudge them onto the new protocol. Presence of startAfter now switches the replay to every record; the value stays ignored. Also document the bare ActionHistoryFilter export.
A throwing entry listener on the shared client store broke the fan-out for every other consumer; guard each listener and log. On the server, a subscriber that fails mid-replay now rejects the subscribe call (the client's error signal) instead of silently returning a dead subscription.
A failed non-first page previously only console.error'd, leaving the Load older button looking inert. Track loadMoreFailed in the hook (first-load failures still own status: 'error') and render an inline retry row in the Activity history view. The cursor is untouched on failure, so retry re-requests the same page.
The header treated the shared subscription's 'error' status as all-clear
('Nothing is waiting on you.'). Thread isError into ActivityNotifications
and render the failure copy instead; pendings gathered before the failure
still render via the non-empty branch.
…dicate deliverMessage re-implemented #getChatMessageForClient's action-log lookup and had already diverged from it (the helper unconditionally hydrates attachments); route it through the helper. The helper's body has no awaits, so it drops async and deliverMessage issues subscriber.message() synchronously, keeping the documented messages-before-metadata delivery order intact across the metadata()/deleted() callbacks (the client's provisional-stream mop-up on activeAgent unset relies on it). listActions' type filter moves to matchesActionHistoryFilter in workshop-shared so the server page filter and the client live-merge can't drift.
…sumers - ActionsState now exposes one sorted readonly pending array instead of pendingById + entriesById; the byte-identical createdAt||id sorts in Activity and ActivityNotifications collapse into the store's commit(), and GadgetEditor's hookSignature memo (the sole entriesById consumer, one full-log Map clone per committed frame) becomes a useActionEntries fold over just the bindHook entries. - ActivityNotifications subscribes via useActions itself (the store is ref-counted per stub), dropping the pendingById/isChecking/isError prop drilling and restoring the 3-state status union. - useActionHistory drops the never-read 'idle' status; Activity's initial-loading branch keys off status === 'loading', so hasMore means only what the server said. The duplicated Load-older button is one local component. - useActions/useActionHistory tests share one harness (entry factory, fake overseer, act/rAF root management) in action-test-harness.ts instead of ~70 drifting lines.
Indexes are only maintained by write-time subscribers, so an index declared over pre-existing records starts empty -- and the first update to such a record throws on the index's remove. rebuild() clears the index's storage and re-derives it from the collection, giving migrations a way to backfill a newly declared index.
Declare a sparse pendingByGatekeeper index on the actions collection (pending records only, keyed by gatekeeper) and point the auto-approval drain at it: the drain reads the gatekeeper's pendings off the index (materialized before applying; the index yields in ascending id order) instead of paging the whole log. DRAIN_PAGE_SIZE, the cursor loop, and the between-page yields go away. Storage schema version 3 backfills the index via rebuild() -- mandatory, not an optimization: indexes are only maintained at write time, so resolving a pre-index pending record would otherwise throw on the index update. The migration is synchronous and guarded on version == 2, chaining after the git-storage migration inside its blockConcurrencyWhile when that one is still pending; never-initialized DOs stay write-free. Tests move onto shared fixtures (__tests__/fixtures.ts): one production- mirroring actions schema, one putAction, and one fake-overseer builder replacing the drifted copies in action-log-pagination, auto-approval, and overseer-hooks; auto-approval gains pre-index-backfill coverage, and git-migration-do covers both constructor triggers (v1 ladder through to 3, and the v2 backfill alone) over real DO storage.
…"pending"}) Extend ActionHistoryFilter with a "pending" member instead of minting a new RPC: the same ActionHistoryPage/beforeId paging applies, the use-role inert path already answers any filter with an empty terminal page, and the capnweb validator regenerates from the widened union. matchesActionHistoryFilter becomes state-aware -- "pending" matches pending records of any type; the resolved filters now exclude pending themselves, so listActions' scan path drops its explicit pending skip in favor of the shared predicate. The pending branch is served from the pendingByGatekeeper index: materialize the (gatekeeper-grouped) pending set, cut the id-descending page below beforeId. O(pending) per page -- documented as wire-bounded but not server-work-bounded until typed-storage grows per-group range reads for a k-way merge (noted as a follow-up, not built here). Pages reflect call-time state: records resolved since an earlier page stop appearing; a subscription carries their resolutions. Docs spell out the query-for-state half of the contract.
The shared action store now initiates subscribeToActions (live deltas; no
startAfter), then concurrently pages listActions({filter: 'pending'}) --
capnweb e-order registers the subscriber server-side before the first page
reads, so pages snapshot call-time state and the subscription carries
everything after. Settledness ('checking' -> 'ready') moves from the
subscribe call resolving to the last pending page loading; a subscribe
rejection downgrades even an already-'ready' store, since a dead live stream
must not present as settled.
Pages fold live-wins via a new liveSeenIds set: any id the subscription has
delivered skips the page fold, so a live resolution is never regressed by its
stale page copy. Records created mid-paging are live-only (ids above page 1's
snapshot bound). entryListeners fan out live entries only; paged records are
not entries.
Compatible with both server behaviors: while the server still replays
pendings through entry(), those mark liveSeenIds and the pages dedupe against
them -- this lands before the server goes live-only.
useActionHistory drops its explicit pending skip (the shared predicate now
excludes pending from resolved filters). Test harness routes the store's
pending queries onto their own queue and logs initiation order.
Delete the pending replay: the subscription now registers and fires ready()
immediately; clients query the current pending set via
listActions({filter: "pending"}) after initiating the subscribe (capnweb
e-order makes the pair gapless -- the contract subscribeToChat already
documents). This removes the unpaced entry() replay push entirely: initial
state now flows over a pull-paged, client-clocked read.
The deprecated startAfter full-replay block stays for pre-deploy clients
(PENDING_SCAN_PAGE_SIZE is now legacy-only, deleted with it); ready() is
marked deprecated -- it only ever ended the replay.
The frontend store already speaks subscribe-then-page (previous commit), so
nothing client-side changes behavior here.
eef947d to
dee1f52
Compare
- auto-approval: drop the no-op throughId bound (the pending snapshot and
nextActionId read are one synchronous block, so the filter never drops
anything); AutoApprovalStorage no longer needs nextActionId
- backend fixtures: makeActionStorage delegates to makeOverseerStorage so the
action suites validate the shipped schema instead of a copy; putAction takes
the minimal structural storage type
- useActions: delete liveSeenIds (stagedEntries carries the same ids), share a
resetSession() helper, skip the re-sort/re-render for entries that never
touch the pending set, drop the Date clones in the commit comparator
- useActionHistory: collapse the View/session mirror to {byId, error} state
with status/hasMore/isLoadingMore derived from the session ref; public
return shape unchanged
- Activity: one ActivityNotice component for the copied notice blocks,
LoadOlderButton grows a label prop (Retry buttons regain their disabled
guard), pending-status copy shared with ActivityNotifications
- typed-storage tests: hoist the duplicated INDEXED_SCHEMA/PLAIN_SCHEMA
literals to file scope
- action-test-harness: one parkedQueue() helper for the history and
pending-query queues (makeOverseer API unchanged)
- ChatInterface: shared getCachedActionMessage() for the four cached action
message lookups
Preview:
|
- useActions: a subscribe failure is sticky -- pending pages draining afterwards no longer upgrade a store with a dead live stream back to 'ready'; test for the failure-before-pages ordering - typed-storage: rebuild() buffers its record scan one page at a time instead of materializing the whole collection, keeping the v2->v3 pending-index migration memory-bounded while staying atomic inside the caller's transactionSync; multi-page backfill test - ChatInterface: the reconnect action-card sweep fetches at most 4 cards concurrently instead of firing every stale card's getChatMessage at once - overseer: the deprecated startAfter replay awaits each page's delivery, bounding outstanding callbacks and rejecting the subscribe call (before ready()) on any delivery failure, the final page's included - useActionHistory: document why the live-merge drop guards can't lose an update (listActions snapshots and responds in one DO turn, so conflicting entries always arrive after the page they'd race with)
| * | ||
| * A page reflects call-time state: a record resolved between pages stops appearing in later | ||
| * "pending" pages (and starts appearing in resolved ones); an action subscription carries its | ||
| * resolution as a live update. |
There was a problem hiding this comment.
This paragraph seems to be stating the obvious, albeit in Claudish so it's hard to tell. I'd delete it.
| * | |
| * A page reflects call-time state: a record resolved between pages stops appearing in later | |
| * "pending" pages (and starts appearing in resolved ones); an action subscription carries its | |
| * resolution as a live update. |
| * The subscription delivers live deltas only — nothing pre-existing is replayed. Query for | ||
| * state, subscribe for deltas: fetch the current pending set via | ||
| * listActions({filter: "pending"}) and resolved history via the other filters. As with | ||
| * subscribeToChat(), initiate the subscribe call before those reads — there is no need to | ||
| * await its return, only to start it first — so nothing can slip between the snapshot the | ||
| * pages reflect and the stream. | ||
| * | ||
| * `startAfter` is deprecated: its presence makes the server replay every record through the | ||
| * subscriber before ready(), since pre-deploy clients derive their entire history view from | ||
| * replay. The value itself is ignored. New clients must omit it. | ||
| * TODO: Delete it once pre-deploy clients have cycled out. |
There was a problem hiding this comment.
Please don't deprecate startAfter.
| * The subscription delivers live deltas only — nothing pre-existing is replayed. Query for | |
| * state, subscribe for deltas: fetch the current pending set via | |
| * listActions({filter: "pending"}) and resolved history via the other filters. As with | |
| * subscribeToChat(), initiate the subscribe call before those reads — there is no need to | |
| * await its return, only to start it first — so nothing can slip between the snapshot the | |
| * pages reflect and the stream. | |
| * | |
| * `startAfter` is deprecated: its presence makes the server replay every record through the | |
| * subscriber before ready(), since pre-deploy clients derive their entire history view from | |
| * replay. The value itself is ignored. New clients must omit it. | |
| * TODO: Delete it once pre-deploy clients have cycled out. | |
| * The `startAfter` parameter is intended to be used when resubscribing after a disconnect: | |
| * specify the time of the last action seen, in order to ensure no actions were missed during | |
| * the disconnect. If not specified, the subscription starts from the current time. | |
| * | |
| * Do NOT use `startAfter` as a way to enumerate historical data. Use `listActions()` instead. | |
| * To ensure no holes between a subscription and historical data, call `subscribeToActions()` | |
| * immediately before `listActions()`, similar to `subscribeToChat()`. |
|
|
||
| /** | ||
| * Filter for listActions(): "pending" for currently-pending records (of any type), or a resolved | ||
| * view — one specific record type, or "all" for every resolved record. |
There was a problem hiding this comment.
Wait, so there's no way to query both pending and non-pending at the same time? Why?
| /** Matching records, descending id (creation order, newest first). May be short or empty while | ||
| * older history remains — absence of `nextBeforeId` is the only terminator. */ |
There was a problem hiding this comment.
Claudish.
| /** Matching records, descending id (creation order, newest first). May be short or empty while | |
| * older history remains — absence of `nextBeforeId` is the only terminator. */ | |
| /** | |
| * Matching records, descending id (creation order, newest first). | |
| * | |
| * Note that an empty list does NOT mean that there are no more pages. It may be that the server | |
| * reached the scan limit for one page without finding any matches. Only the absence of | |
| * `nextBeforeId` indicates that there are no more pages. | |
| */ |
| * @deprecated Fires immediately after a modern (live-only) subscribe registers — it only ever | ||
| * marked the end of the replay, which now exists solely for the deprecated startAfter path. | ||
| * Dies with that path; clients should treat it as a no-op. |
There was a problem hiding this comment.
| * @deprecated Fires immediately after a modern (live-only) subscribe registers — it only ever | |
| * marked the end of the replay, which now exists solely for the deprecated startAfter path. | |
| * Dies with that path; clients should treat it as a no-op. | |
| * @deprecated Fires after the subscription has caught up to the current time. However, this is | |
| * only a useful signal when a subscription is being used to enumerate past actions using a | |
| * distant-past `startAfter`. This is not the correct way to use `subscribeToActions()`; use | |
| * `listActions()` instead. |
| let recordTimestamp = (appliedAt ?? record.createdAt).valueOf(); | ||
| if (recordTimestamp > startAfterTimestamp) { | ||
| subscriber.entry(actionRecordToLog(record)).catch(unsubscribe); | ||
| // DEPRECATED: pre-deploy clients pass startAfter and build their entire history view from |
There was a problem hiding this comment.
Revert these changes, keep startAfter the way it is.
| let result: ActionLogEntry[] = []; | ||
| for (let record of this.impl.storage.actions.list()) { | ||
| result.push(actionRecordToLog(record)); | ||
| async listActions(options?: {beforeId?: number, filter?: ActionHistoryFilter}) |
There was a problem hiding this comment.
This implementation seems pretty suboptimal, especially for "pending".
There are some really pathological cases here: Most workspaces don't add new hooks often, so if you go to the "activity" view and click "hooks", you almost always end up doing a full table scan. Sure, there is a limit on scan, but then this returns no results, which just causes the client to request the next page -- so you end up doing a full scan anyway, just split across several round trips, making it even slower.
We could fix all of this and simplify the code by adding another non-unique index on ActionHistoryFilter. Note that an index function is allowed to return an array of keys, so for a pending action, you could return ["pending", "action"] to appear in both indexes.
| let self = this; | ||
| function deliverMessage(record: AiChatMessage) { | ||
| subscriber.message(self.impl.hydrateChatMessageForClient(record)).catch(unsubscribe); | ||
| subscriber.message(self.#getChatMessageForClient(record)).catch(unsubscribe); |
There was a problem hiding this comment.
I guess this is a bugfix, made necessary since actions can't be cross-referenced from the action subscription anymore?
| // Synchronous so subscribeToChat's deliverMessage can issue message() in the same turn as the | ||
| // metadata()/deleted() calls around it, preserving cross-callback delivery order. |
There was a problem hiding this comment.
| // Synchronous so subscribeToChat's deliverMessage can issue message() in the same turn as the | |
| // metadata()/deleted() calls around it, preserving cross-callback delivery order. |
| } | ||
| async listActions(): Promise<ActionLogEntry[]> { this.#deny(); } | ||
| // Inert (see class comment): the editor pages the action log speculatively before it knows its | ||
| // role, so answer with an empty terminal page instead of denying. |
There was a problem hiding this comment.
Why is the editor loading the log speculatively so early? People almost never look at the activity log. Wouldn't it be better not to try to fetch it at all until someone opens the activity view?
Reading the action log today means reading all of it: opening a workspace replays every record ever written to each client, the Activity pane holds the whole log in React state, and the auto-approval drain materializes the full table per run. Long-lived workspaces pay for this on every open, and the cost only grows.
This branch bounds every read path and splits the protocol into query-for-state / subscribe-for-deltas:
subscribeToActionsdelivers live updates only. Clients fetch the current pending set throughlistActions({filter: "pending"})after initiating the subscribe; capnweb e-order makes the pair gapless. The unpaced replay push is gone; initial state flows over pull-paged, client-clocked reads.listActionsRPC pages history newest-first: resolved records under a raw-scan cap, so a log buried in resolved records returns short pages with a cursor instead of stalling the DO, and pending records off a new sparse pending-by-gatekeeper indexAutoApprovalDrainerreads its gatekeeper's pendings off the index instead of scanning the log.startAfterget a paced full replayTested with unit suites for the subscription, pagination, migration-backfill, and drain paths, an integration smoke over the paged RPCs, and hook tests for the new frontend state.