fix(libs): janitor live-run guard, Container dispose-on-replace, vector default parity, StreamPump cancel#604
Closed
sroussey wants to merge 14 commits into
Closed
fix(libs): janitor live-run guard, Container dispose-on-replace, vector default parity, StreamPump cancel#604sroussey wants to merge 14 commits into
sroussey wants to merge 14 commits into
Conversation
…ask-graph
Addresses correctness, DX, and simplicity findings from a subsystem review
of the core execution stack (util -> storage -> job-queue -> task-graph),
each verified against source and covered by tests.
util:
- BaseError now extends native Error (restores `instanceof Error`, adds cause
support) so the library's own errors get full diagnostics/telemetry
- EventEmitter: fix once/removeAllListeners re-entrancy during dispatch
- Container: honor re-register after a singleton is instantiated; isolate
child-container disposal from parent-owned instances
- SchemaUtils: type-check optional shared object props; compare the full
multi-segment format narrowing (not just the first segment)
- graph: visited-set in canReachFrom (no stack overflow on cycles); deep-copy
state in fromDirectedGraph; dedup edges by identity; typed GraphInvariantError
- DirectedGraph.addEdge: drop the boolean param (moved to a protected method)
storage:
- Atomic putBulk + rollback in the in-memory tabular base; route post-commit
mutation emits through safeEmit and correct its doc
- FsFolder KV: idempotent delete (swallow ENOENT); typed StorageUnsupportedError
for getAll/size instead of a bare Error
- mapPostgresType: widen signed/large-maximum integers to BIGINT
- typedArrayCtors: add Float16Array to match the documented vector set
- CachedTabularStorage: surface cache-init failures; serialize subscription
cache updates instead of floating async work
- vector: default scoreThreshold to -Infinity; null-safe metadata filter;
emit the declared similaritySearch event
job-queue:
- Log errors in the worker main loop instead of swallowing them
- TelemetryQueueStorage: pass through optional findActiveByFingerprint so the
decorator doesn't silently degrade dedup to an O(n) scan
- Unify IClaim.ack(undefined) semantics (overwrite with null) across backends
- ConcurrencyLimiter: guard complete() against foreign tokens; bound the
processingTimes buffer; per-instance scan-exhaustion warning
task-graph:
- ConditionalBuilder: wire the conditional's input from the preceding task
(Blocker: predicate no longer sees {}); guard ambiguous two-arm continuation
- Derive the run id from caller config so run-scoped job cancellation matches
- serialGraph: correct source/target port order
- TaskGraphRunner: guard preview against a concurrent run; restore the registry
after preview
- StreamPump: close the stream and detach listeners on error/abort; guard
cross-graph emits against throwing consumers
- IteratorTask/WhileTask: throw TaskAbortedError on mid-iteration abort instead
of returning partial results as COMPLETED
- CacheCoordinator: treat an undecodable cache entry as a miss, not a hard fail
- Workflow: emit the declared abort event; public addTask throws on
auto-connect failure instead of silently dropping the task
Plus assorted minor fixes and regression tests across all four packages.
Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01N57Pgeeqgsf4dYBEdijLdq
…er backends
Follow-up to the core review fixes, applying the same storage contracts to the
remaining backends so behavior no longer diverges when swapping implementations.
indexeddb:
- IndexedDbVectorStorage: default scoreThreshold to -Infinity (stop silently
dropping negatively-correlated hits), coalesce a present-but-null metadata
value to {} under a filter (no throw), and emit the declared
similaritySearch event
- IndexedDbQueueStorage: complete()/saveProgress() silently no-op on a missing
job instead of throwing, matching the documented IQueueStorage contract
(lease-expiry races can legitimately remove the row between claim and write)
sqlite / postgres / supabase tabular backends:
- Route post-commit mutation/read emits through safeEmit so a throwing
subscriber cannot turn an already-committed write into a rejection
- Emit the rollback event on putBulk failure (putBulk is already atomic via the
backend transaction / server-side upsert, so no rows persist: ids: [])
Adds IndexedDbVectorStorage result-behavior tests mirroring the in-memory suite.
Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01N57Pgeeqgsf4dYBEdijLdq
…able
The run-private output cache namespaced entries by prepending `__run:${runId}::`
to the `taskType` axis and cleaned up via full-table scans + per-row deletes
(`deleteByTaskTypePrefix` and friends), which ran on every successful run
(`RunPrivateCacheRepo.clearRun`) and in the janitor sweep — O(all cached rows).
The run-private and deterministic caches are now separate CacheRegistry slots
(separate tables), so the prefix only distinguished runs within the private
table. Give the private cache its own shape instead:
- New RunPrivateTaskOutputSchema: a first-class `runId` column with a
runId-leading primary key [runId, key, taskType]. clearRun and the janitor
sweep become indexed deleteSearch({ runId }) / deleteSearch({ createdAt: "<" })
rather than table scans, and two runs with identical (taskType, inputs) no
longer collide.
- RunPrivateTaskOutputRepository implements run-scoped saveOutputForRun /
getOutputForRun / deleteRun / deleteRunOlderThan / sizeForRun; the base
TaskOutputRepository declares these (default-throw) in place of the removed
taskType-prefix methods, and the deterministic TaskOutputTabularRepository
drops the three scan methods.
- RunPrivateCacheRepo delegates to the backing's run-scoped methods (no prefix);
CacheJanitor.sweepStaleRunPrivate is an indexed createdAt delete over the
dedicated table. Stale "deterministic cache entries" docs corrected.
- Shared output encode/decode extracted to taskOutputCodec.
No backward compatibility with the old prefixed rows: the private cache is
ephemeral and ages out, so consumers point the private slot at a
RunPrivate* repository (RunPrivateInMemoryTaskOutputRepository binding added)
and clear any old prefixed store on upgrade.
Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01N57Pgeeqgsf4dYBEdijLdq
From a max-effort review of the previous commit:
- taskOutputCodec: the uncompressed decode path called `.toString()` on the raw
value, which yields comma-separated byte numbers (not JSON) when a durable
backend (SQL/IndexedDB) hands the blob back as a Uint8Array — JSON.parse then
crashed. Decode the bytes as UTF-8 instead; shape-normalization shared with the
compressed path. Adds a round-trip + plain-Uint8Array regression test.
- RunPrivateTaskOutputRepository.sizeForRun: use storage.count({ runId }) instead
of query({ runId }).length, which materialized and decoded every row's blob
just to take a length.
- CacheJanitor: type `privateBacking` as RunPrivateTaskOutputRepository (not the
base TaskOutputRepository) so a per-run RunPrivateCacheRepo wrapper — whose
clearOlderThan is scoped to one run — can't be passed by mistake, which would
leave other runs' stale rows un-swept.
- README / EXECUTION_MODEL: the cache-wiring example wired a TaskOutputTabularRepository
as the private slot (now throws saveOutputForRun) and described the removed
`__run:` key prefix. Updated to RunPrivateTaskOutputRepository + the runId-column
model, and corrected the janitor's "deterministic never affected" note.
Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01N57Pgeeqgsf4dYBEdijLdq
- taskOutputCodec.encodeTaskOutput: use TextEncoder instead of Buffer.from for the uncompressed path so it works in browser builds (Buffer is undefined there); mirrors the TextDecoder decode fix. - BaseError: derive `name` from the subclass's OWN `static type` (hasOwnProperty) and otherwise fall back to the runtime constructor name. The inherited `BaseError.type` made the documented constructor-name fallback unreachable, so a subclass that forgot to declare `type` was mislabeled "BaseError". - Dataflow: correct the DataflowArrow comment — ids/ports may contain any character except the structural `[` `]` delimiters (the old comment claimed no restriction). Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01N57Pgeeqgsf4dYBEdijLdq
…ht cache-row deletion CacheJanitor.sweepStaleRunPrivate previously called RunPrivateTaskOutputRepository.clearOlderThan unconditionally, which deleted every row older than the cutoff — including rows belonging to long-running in-flight runs (createdAt is the row write time, not run liveness). Add a liveRunIds: () => Iterable<string> snapshot accessor on CacheJanitor. Default to () => [] with a one-time console.warn since defaulting is unsafe when any run is ever concurrent with the sweep. Plumb the snapshot Set through to clearOlderThan, which now takes excludeRunIds and, when present, enumerates distinct old runIds and calls deleteRunOlderThan per runId not in the exclude set (ITabularStorage has no NOT IN operator).
…ment Container.register and Container.registerInstance overwrote a cached singleton without invoking its disposer protocol. A DB connection, file handle, or event subscriber registered then re-registered would leak the original. Extract the disposer chain into invokeDisposer (the protocol probe) and a disposeService eviction wrapper that catches errors with console.warn so a buggy disposer cannot block re-registration. dispose() retains its own try/catch + AggregateError collection by calling invokeDisposer directly. Inherited (parent-owned) instances are skipped — child cannot dispose an instance the parent still hands out.
…ld to 0 (match SQL backends) InMemoryVectorStorage and IndexedDbVectorStorage defaulted scoreThreshold to -Infinity, while every SQL backend (Sqlite / Postgres / Supabase / SqliteAiVectorStorage) defaults to 0. Identical app code returned a different result set depending on the backing — a portability foot-gun. Switch in-memory + IndexedDB defaults to 0. Callers that want every result regardless of correlation pass scoreThreshold: -Infinity explicitly (which is exactly what the SQL backends already required). Inverted the two validation tests that previously asserted the negative-correlation behaviour.
…mFromTaskEvents createStreamFromTaskEvents wired stream_chunk / stream_end / status listeners onto the source task but only released them on a normal stream_end or terminal task status. A consumer that called reader.cancel() mid-stream (downstream abort, consumer-side limit hit) left every listener attached: the task kept emitting into a closed controller and the listeners pinned GC. Hoist the shared cleanup closure to underlying-source scope so the ReadableStream cancel handler can call it. cleanup is already idempotent via the closed flag, so concurrent cancel + stream_end is safe.
…derThan
`RunPrivateTaskOutputRepository.clearOlderThan` invoked
`this.storage.search(...)`, but `ITabularStorage` has no `search` method — the
non-empty-`excludeRunIds` branch would throw `TypeError` at runtime. Swap in
`this.storage.query({...})`, which has the same criteria shape.
Add a smoke test `clearOlderThan with a non-empty excludeRunIds does not throw`
so the regression stays regressed.
… fast-path The optional `liveRunIds` callback let callers silently opt into a sweep that could reap in-flight runs' cache rows — a one-time `console.warn` isn't sufficient. Make the option required so the empty-list case is a conscious choice at every call site. Also drop the `excludeRunIds.size === 0` early-return branch in `RunPrivateTaskOutputRepository.clearOlderThan`. The distinct-runId loop below is correct for the empty-set case: it iterates every stale runId and calls `deleteRunOlderThan`, so no rows are missed. Removing the fast-path centralizes the eviction path so the streaming rewrite in the follow-up commit is the sole implementation. Existing tests updated to pass `liveRunIds: () => []` explicitly. New tests: - `omitting liveRunIds is a compile-time error` (`@ts-expect-error`). - `empty liveRunIds still respects the per-run indexed delete path` — seeds 3 stale runs + 1 live-but-excluded run and asserts the excluded rows survive. README example updated to match the new signature.
…egisterInstance `register()` and `registerInstance()` fired eviction disposers with `void`, so `dispose()` (and the caller) had no way to observe when the released resource actually shut down. A fast re-register / dispose sequence could clear container state while the async disposer was still holding a DB connection open. Track in-flight eviction disposals in a `pendingDisposals` map, expose `awaitReplacement(token)` / `awaitRegistrations()`, and drain the map at the top of `dispose()`. Entries are identity-guarded via `.finally()` so a subsequent replacement's promise doesn't get erased. Also: `registerInstance()` now clears any previously-registered factory for the token. Leaving the stale factory behind meant a later `remove()` + `get()` could resurrect the factory-built object instead of failing loudly. New tests: - `register schedules eviction dispose and awaitReplacement resolves after it completes` - `container.dispose() drains pending disposals from prior register replacements` - `registerInstance eviction is drained by awaitReplacement` - `registerInstance clears any previously-registered factory for the token`
The previous implementation of `RunPrivateTaskOutputRepository.clearOlderThan`
issued `storage.query({ createdAt: { < } })` for the ENTIRE stale row set in
one call, then walked the buffer to extract distinct runIds. On a table with
millions of stale rows this materializes an O(row-count) buffer of Entity
objects (including the blob column) — a memory hazard proportional to how
stale-friendly the caller's cutoff is.
Rewrite the sweep to iterate the stale set via cursor-paginated `queryPage`
in bounded 500-row batches. `seenRunIds` is bounded by distinct runId count
(typically much smaller than row count), and each page carries a `limit` so
SQL backends push both the WHERE and the keyset predicate into the SELECT.
Default ordering is PK ASC (runId-leading), which groups a run's rows
contiguously and lets distinct runIds accumulate quickly.
Also introduce a private `deleteRunOlderThanAt(runId, cutoff)` helper so the
sweep can dispatch per-run deletes against a single pre-computed cutoff
without triggering the per-run `output_pruned` emit. The public
`deleteRunOlderThan` still emits per call; the all-runs sweep emits exactly
once at the end.
New tests:
- `sweepStaleRunPrivate does not materialize the full stale row set at once`
seeds 5000 stale rows across 3 runs and asserts `queryPage` is called ≥10
times with a bounded `limit` on every call.
- `clearOlderThan preserves live runs even when they have stale rows` (in a
new `RunPrivateTaskOutputRepository.test.ts`) verifies the excludeRunIds
contract when the excluded run's rows all predate the cutoff.
Contributor
There was a problem hiding this comment.
Pull request overview
This PR stacks follow-up correctness and parity fixes across the core execution stack (task-graph/util/storage/job-queue/providers), with accompanying regression tests. It focuses on run-private cache safety, service/container lifecycle cleanup, vector-search default consistency across backends, and stream cancellation listener cleanup.
Changes:
- Harden run-private cache infrastructure: live-run-safe janitor sweep, runId-as-column storage model, and output codec normalization.
- Make storage/backend event emission resilient to throwing listeners via
safeEmit, plus add rollback signaling on bulk failures. - Improve operational correctness in streaming/cancellation, worker abort dedupe windows, and queue worker/limiter behavior, with targeted regression tests.
Reviewed changes
Copilot reviewed 119 out of 119 changed files in this pull request and generated 4 comments.
Show a summary per file
| File | Description |
|---|---|
| providers/supabase/src/storage/SupabaseTabularStorage.ts | Use safeEmit and emit rollback on bulk upsert failure |
| providers/sqlite/src/storage/SqliteTabularStorage.ts | Use safeEmit; emit rollback when transaction throws |
| providers/postgres/src/storage/PostgresTabularStorage.ts | Use safeEmit; emit rollback on transaction rollback |
| packages/util/src/worker/WorkerServerBase.ts | Document cooperative cancellation; align completed-id TTL with abort TTL |
| packages/util/src/worker/Worker.node.ts | Forward base constructor options to WorkerServerBase |
| packages/util/src/worker/Worker.bun.ts | Forward base constructor options to WorkerServerBase |
| packages/util/src/worker/Worker.browser.ts | Forward base constructor options to WorkerServerBase |
| packages/util/src/utilities/BaseError.ts | Make BaseError extend native Error (cause + prototype fixups) |
| packages/util/src/resource/ResourceScope.ts | Log disposer failures while keeping best-effort semantics |
| packages/util/src/json-schema/SchemaValidation.ts | Improve format docs; reject array-as-schema objects |
| packages/util/src/json-schema/SchemaUtils.ts | Fix multi-segment format narrowing and optional-prop compatibility |
| packages/util/src/json-schema/parsePartialJson.ts | Better handling of first-property incomplete bare tokens |
| packages/util/src/graph/graph.ts | Use forEach for adjacency expansion; dedup edges by identity; fix edge-removed emits |
| packages/util/src/graph/errors.ts | Add typed GraphInvariantError |
| packages/util/src/graph/directedGraph.ts | Prevent canReachFrom recursion overflow via visited set |
| packages/util/src/graph/directedAcyclicGraph.ts | Deep-copy from DirectedGraph; avoid redundant cycle checks; throw GraphInvariantError |
| packages/util/src/events/EventEmitter.ts | Fix once/removeAllListeners re-entrancy semantics |
| packages/util/src/di/ServiceRegistry.ts | Make container readonly |
| packages/test/src/test/vector/InMemoryVectorStorage.validation.test.ts | Add tests for default scoreThreshold behavior + events |
| packages/test/src/test/vector/IndexedDbVectorStorage.validation.test.ts | Add tests for default scoreThreshold behavior + events |
| packages/test/src/test/util/WorkerServerBase.race.test.ts | Regression test for abort/call dedupe-window terminal replies |
| packages/test/src/test/util/SchemaValidation.test.ts | Test array-valued schema property error reporting |
| packages/test/src/test/util/SchemaUtils.test.ts | Expand semantic-compat tests (optional props, tuples, multi-segment formats) |
| packages/test/src/test/util/parsePartialJson.test.ts | New tests for first-property incomplete keyword/number streaming |
| packages/test/src/test/util/graph.test.ts | Tests for edge-removed emission semantics |
| packages/test/src/test/util/EventEmitter.test.ts | Tests for re-entrant once and removeAllListeners behavior |
| packages/test/src/test/util/directedGraph.test.ts | Tests for canReachFrom termination on cycles/diamonds |
| packages/test/src/test/util/directedAcyclicGraph.test.ts | Test DAG independence from source graph backing state |
| packages/test/src/test/util/BaseError.test.ts | Tests for instanceof Error and cause propagation |
| packages/test/src/test/task/IteratorTask.test.ts | Regression test: mid-flight abort rejects (no partial COMPLETED result) |
| packages/test/src/test/task/ConditionalTaskCondition.test.ts | Regression test: non-numeric ordering operators evaluate false |
| packages/test/src/test/task-graph/Workflow.test.ts | Fluent addTask throws on auto-connect failure; abort emits abort not error |
| packages/test/src/test/task-graph/TaskGraph.test.ts | Fix serialGraph port direction expectation; verify dataflow ids |
| packages/test/src/test/task-graph/runner-internals.test.ts | Test StreamPump listener cleanup on reader.cancel() |
| packages/test/src/test/task-graph/ResourceScope.test.ts | Test disposeAll best-effort behavior with rejecting disposer |
| packages/test/src/test/task-graph/ConditionalBuilder.test.ts | Tests for mid-chain .if wiring and ambiguous continuation behavior |
| packages/test/src/test/task-graph-output-cache/RunPrivateTaskOutputRepository.test.ts | New test: preserve live runs during age sweep |
| packages/test/src/test/task-graph-cache/TaskGraphRunnerPrivate.test.ts | Swap private repo binding to run-private backing |
| packages/test/src/test/task-graph-cache/TaskGraphRunnerDurabilityWarning.test.ts | Update to run-private backing; adjust expectations |
| packages/test/src/test/task-graph-cache/TaskGraphRunnerCleanup.test.ts | Update to run-private backing; cleanup behavior tests |
| packages/test/src/test/task-graph-cache/RunPrivateOutputCodec.test.ts | New tests for output codec + Uint8Array round-trip shapes |
| packages/test/src/test/task-graph-cache/RunPrivateCacheRepo.test.ts | Update semantics away from taskType-prefix namespacing |
| packages/test/src/test/task-graph-cache/RunPrivateCacheKeyFallback.test.ts | Update repo type usage + run-private keying |
| packages/test/src/test/task-graph-cache/PrivateRequiresRunId.test.ts | Update assertions for runId column model |
| packages/test/src/test/task-graph-cache/AbstractRepoCapabilities.test.ts | Update tests for default-throwing run-scoped methods |
| packages/test/src/test/storage-util/sqlTypeMapping.test.ts | New tests for BIGINT mapping + Float16Array ctor presence |
| packages/test/src/test/storage-tabular/InMemoryTabularStorage.test.ts | Tests for atomic putBulk rollback + getOffsetPage validation |
| packages/test/src/test/storage-tabular/CachedTabularStorage.integration.test.ts | Warm-up failures now surface; init remains retryable |
| packages/test/src/test/storage-kv/FsFolderKvRepository.integration.test.ts | Tests for idempotent delete + typed unsupported errors |
| packages/test/src/test/job-queue/TelemetryQueueStorage.test.ts | Ensure optional findActiveByFingerprint delegates correctly |
| packages/test/src/test/job-queue/Limiters.test.ts | Tests for ConcurrencyLimiter token guarding |
| packages/test/src/test/job-queue/JobQueueWorker.test.ts | Test server job_error forwards errorCode |
| packages/test/src/binding/RunPrivateInMemoryTaskOutputRepository.ts | New in-memory binding for run-private cache repository |
| packages/task-graph/src/task/WhileTask.ts | Abort becomes TaskAbortedError; unify condition error wrapping |
| packages/task-graph/src/task/TaskRunner.ts | Document process-global warning set implications |
| packages/task-graph/src/task/Task.ts | Replace console.warn with logger; clarify undefined-merge semantics |
| packages/task-graph/src/task/IteratorTaskRunner.ts | Abort becomes TaskAbortedError (no partial success) |
| packages/task-graph/src/task/iterationSchema.ts | Remove unused alias export |
| packages/task-graph/src/task/FallbackTaskRunner.ts | Clarify event-bridging gap in task-mode fallback |
| packages/task-graph/src/task/FallbackTask.ts | Thread JSON options through toJSON override |
| packages/task-graph/src/task/ConditionUtils.ts | Log-and-false for NaN ordering comparisons |
| packages/task-graph/src/task/ConditionalTask.ts | Document dual output shapes (function branches vs conditionConfig) |
| packages/task-graph/src/task/CacheCoordinator.ts | Treat decode failures as cache misses (log + recompute) |
| packages/task-graph/src/task-graph/WorkflowEventBridge.ts | Update lifecycle docs: beginRun returns teardown token |
| packages/task-graph/src/task-graph/Workflow.ts | Abort emits abort event; fluent addTask throws on auto-connect failure |
| packages/task-graph/src/task-graph/TaskGraphRunner.ts | Thread caller runId into RunContext; prevent preview/run overlap; restore preview registry |
| packages/task-graph/src/task-graph/TaskGraphEvents.ts | Document repeated task_complete emissions across iterations |
| packages/task-graph/src/task-graph/TaskGraph.ts | Optimize getDataflow by parsing target id; fix serialGraph edge port direction |
| packages/task-graph/src/task-graph/StreamPump.ts | Guard graph stream emits; cleanup listeners on cancel() and terminal status |
| packages/task-graph/src/task-graph/RunScheduler.ts | Treat scheduler-iterator throws as graph failures |
| packages/task-graph/src/task-graph/RunContext.ts | Accept caller runId so per-task runnerId matches run-scoped operations |
| packages/task-graph/src/task-graph/LoopBuilderContext.ts | Warn on empty loop body finalize |
| packages/task-graph/src/task-graph/Dataflow.ts | Broaden DataflowArrow parsing to match createId grammar |
| packages/task-graph/src/task-graph/ConditionalBuilder.ts | Wire prior output into conditional input; enforce ambiguous continuation rule |
| packages/task-graph/src/storage/TaskOutputTabularRepository.ts | Use shared task output codec; remove taskType-prefix run-private methods |
| packages/task-graph/src/storage/TaskOutputRepository.ts | Replace prefix-based methods with run-scoped methods (default-throwing) |
| packages/task-graph/src/storage/taskOutputCodec.ts | New codec for compress/uncompress + round-trip shape normalization |
| packages/task-graph/src/storage/RunPrivateTaskOutputSchema.ts | New schema for run-private cache table (runId-leading PK) |
| packages/task-graph/src/storage/ITaskOutputStorage.ts | Clarify blob column runtime shapes and normalization |
| packages/task-graph/src/EXECUTION_MODEL.md | Update private cache model docs (runId as column) |
| packages/task-graph/src/common.ts | Export run-private repository + schema |
| packages/task-graph/src/cache/RunPrivateCacheRepo.ts | Switch from taskType-prefix to run-scoped backing methods |
| packages/task-graph/src/cache/CacheJanitor.ts | Add live-run exclusion; use run-private backing + per-run deletes |
| packages/task-graph/README.md | Update cache setup + janitor usage docs for run-private backing |
| packages/storage/src/vector/README.md | Update vector search option docs (scoreThreshold semantics) |
| packages/storage/src/vector/InMemoryVectorStorage.ts | Default scoreThreshold=0; null-safe metadata; emit similaritySearch; reuse atomicPutBulk |
| packages/storage/src/util/HybridSubscriptionManager.ts | Use logger for BroadcastChannel init failures |
| packages/storage/src/tabular/SharedInMemoryTabularStorage.ts | Use logger; forward events via safeEmit; improve error logging |
| packages/storage/src/tabular/ITabularStorage.ts | Clarify getBulk/get events + rollback support semantics |
| packages/storage/src/tabular/InMemoryTabularStorage.ts | Add atomicPutBulk + rollback; safeEmit for events; validate getOffsetPage |
| packages/storage/src/tabular/HuggingFaceTabularStorage.ts | Use safeEmit for get/query events |
| packages/storage/src/tabular/HttpTabularProxyStorage.ts | Use safeEmit for all emitted events |
| packages/storage/src/tabular/FsFolderTabularStorage.ts | Use safeEmit; make delete idempotent; validate pagination; improve logging |
| packages/storage/src/tabular/CachedTabularStorage.ts | Surface cache warm-up failures; safeEmit forwarding; serialize subscribed cache updates |
| packages/storage/src/tabular/BaseTabularStorage.ts | safeEmit for getBulk |
| packages/storage/src/sql/typedArrayCtors.ts | Add conditional Float16Array support |
| packages/storage/src/sql/mapPostgresType.ts | Map large signed integers to BIGINT correctly |
| packages/storage/src/kv/TelemetryKvStorage.ts | Narrow Key generic constraint to string |
| packages/storage/src/kv/KvViaTabularStorage.ts | safeEmit for KV events; log JSON parse failures; removeAllListeners on destroy |
| packages/storage/src/kv/IKvStorage.ts | Narrow Key generic constraint to string |
| packages/storage/src/kv/FsFolderKvStorage.ts | safeEmit; idempotent delete; typed unsupported errors |
| packages/storage/src/events/safeEmit.ts | Expand rationale and clarify post-commit semantics |
| packages/job-queue/src/queue-storage/wrapQueueStorage.ts | Per-instance warning gate for bounded fingerprint scan exhaustion |
| packages/job-queue/src/queue-storage/TelemetryQueueStorage.ts | Preserve optional findActiveByFingerprint presence semantics |
| packages/job-queue/src/queue-storage/IQueueStorage.ts | Document missing-id no-op contracts for complete/progress; clarify finalize role |
| packages/job-queue/src/queue-storage/InMemoryMessageQueue.ts | Align ack/fail overwrite semantics across backends |
| packages/job-queue/src/limiter/ConcurrencyLimiter.ts | Guard complete() by sentinel token |
| packages/job-queue/src/job/JobQueueWorker.ts | Rate-limit loop error logging; bound processing-time samples |
| packages/job-queue/src/job/JobQueueServer.ts | Forward errorCode in job_error; use logger for deletion/cleanup errors; clarify totalJobs semantics |
| packages/job-queue/src/job/JobQueueClient.ts | Implement true sendBatch via messageQueue.sendBatch; refactor job body builder |
| packages/job-queue/src/job/JobErrorRegistry.ts | Use logger for reconstructor overwrite warning |
| packages/indexeddb/src/storage/IndexedDbVectorStorage.ts | Default scoreThreshold=0; null-safe metadata; emit similaritySearch |
| packages/indexeddb/src/job-queue/IndexedDbQueueStorage.ts | Make complete() missing-row a no-op; progress missing-row no-ops with warning |
| packages/test/src/test/task-graph-cache/CacheJanitor.test.ts | Tests for live-run exclusion + pagination behavior in janitor sweep |
💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.
Comment on lines
+104
to
+106
| getLogger().warn( | ||
| `Non-numeric operand in '${operator}' comparison (field=${JSON.stringify(fieldValue)}, compare=${JSON.stringify(compareValue)}); evaluating as false` | ||
| ); |
| readonly topK?: number; // Number of results (default: 10) | ||
| readonly filter?: Partial<Metadata>; // Filter by metadata fields | ||
| readonly scoreThreshold?: number; // Minimum score 0-1 (default: 0) | ||
| readonly scoreThreshold?: number; // Minimum cosine score in [-1, 1]; default -Infinity (no floor, pure topK) |
Comment on lines
57
to
61
| export interface IKvStorage< | ||
| Key extends string | number = string, | ||
| Key extends string = string, | ||
| Value = any, | ||
| Combined = { key: Key; value: Value }, | ||
| > { |
This was referenced Jul 2, 2026
Merged
Collaborator
Author
|
Superseded by #606, which cherry-picks this PR's four fixes (janitor live-run guard, Generated by Claude Code |
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Stacks on #602. Four independent follow-up fixes.
1. (CRITICAL)
CacheJanitorlive-run guardsweepStaleRunPrivatecalledclearOlderThanunconditionally —createdAtis the row write time, not run liveness, so a long-running in-flight run whose rows predate the cutoff would have its cache deleted out from under it.CacheJanitornow accepts aliveRunIds: () => Iterable<string>snapshot (default() => []+ one-timeconsole.warn) and passes the live set intoclearOlderThan(olderThanInMs, excludeRunIds). The repo enumerates distinct old runIds and callsdeleteRunOlderThanper runId not in the exclude set (ITabularStoragehas noNOT INoperator).2. (HIGH)
Container.registerdispose-on-replaceregister/registerInstanceoverwrote a cached singleton without invoking its disposer. A DB connection, file handle, or event subscriber re-registered would leak the original. Extract the disposer chain intoinvokeDisposer(protocol probe) +disposeService(eviction wrapper that catches errors withconsole.warnso a buggy disposer cannot block re-registration).dispose()retains itsAggregateErrorcollection by callinginvokeDisposerdirectly. Inherited (parent-owned) instances are skipped.3. (HIGH) Vector default
scoreThresholdparityInMemoryVectorStorageandIndexedDbVectorStoragedefaultedscoreThresholdto-Infinity; every SQL backend (Sqlite / Postgres / Supabase / SqliteAi) defaults to0. Identical app code returned different result sets depending on the backing — a portability foot-gun. Switched in-memory + IndexedDB defaults to0. Callers that want every result regardless of correlation passscoreThreshold: -Infinityexplicitly. Inverted the two existing validation tests asserting the old behaviour.4. (HIGH)
StreamPump.createStreamFromTaskEventscancel handlerThe wired stream_chunk / stream_end / status listeners only released on a normal stream_end or terminal task status. A consumer calling
reader.cancel()mid-stream left every listener attached: the task kept emitting into a closed controller and the listeners pinned GC. Hoistcleanupout ofstart()so thecancelhandler can call it.cleanupis idempotent.Verification
bun run build:packagesclean (incl. types).bun scripts/test.ts graph vitest— 736 passed.bun scripts/test.ts util vitest— 648 passed, 10 skipped.bun scripts/test.ts storage vitest— 1484 passed, 13 skipped.Generated by Claude Code