Delete the TypeScript re-entry guard; state the store-wait rule once
The AsyncLocalStorage guard refused a store's change handler or timer writing back to its cache without the store waiting for it, which only queues, and still missed a wait that escapes the async context. The serial queue and its own-promise chain stay. The storePushThatAwaitsItsOwnCacheIsRefusedInsteadOfHanging test is replaced by storeChangeHandlerWritingBackQueuesAfterTheOperation: a store push emits an event whose handler puts without being awaited; both writes land in order and store and cache agree. The contract's Locking section states the rule for all runtimes: a store call must not wait for a write to its own cache; unawaited write-backs queue. Python refuses a same-thread re-entrant mutation; no runtime detects a wait on another thread or async context. The globals row states the two-step legacy retirement with compensation, and that a crash or double failure leaving both records is refused at the next pull. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
This commit is contained in:
parent
9c3c92634a
commit
47c7111897
3 changed files with 43 additions and 48 deletions
|
|
@ -1,5 +1,3 @@
|
|||
import { AsyncLocalStorage } from "node:async_hooks";
|
||||
|
||||
import { decode, encode } from "@msgpack/msgpack";
|
||||
|
||||
import type {
|
||||
|
|
@ -52,11 +50,6 @@ type HydratedState = {
|
|||
readonly legacyGlobals: Map<string, CultCacheEnvelope>;
|
||||
};
|
||||
|
||||
// The operations running in the current async flow. A queued call from inside one of this cache's
|
||||
// own operations would wait on itself, so it is refused at entry.
|
||||
type RunningOperation = { readonly cache: CultCache; running: boolean };
|
||||
const runningOperations = new AsyncLocalStorage<readonly RunningOperation[]>();
|
||||
|
||||
// Lookups for every held record of one type under candidate accessors, derived before anything is installed.
|
||||
type DerivedTypeLookups = {
|
||||
readonly entryLookups: Array<[HydratedEntry, EntryLookups]>;
|
||||
|
|
@ -170,34 +163,21 @@ export class CultCache {
|
|||
// so validation, the store call and applying to the cache never interleave with another
|
||||
// mutation of this cache. Reads are not queued. Observer delivery is not scheduled here.
|
||||
// The chain advances on its own promise; the caller gets the operation's promise, untouched, so
|
||||
// an unawaited failure is still an unhandled rejection. Stores, decoders, accessors and updaters
|
||||
// must not call a queued method of the cache they are serving: that call throws.
|
||||
// an unawaited failure is still an unhandled rejection. A call made from inside an operation
|
||||
// queues behind it; a store call that waits for that call waits on itself and never finishes.
|
||||
#serial<T>(operation: () => T | Promise<T>): Promise<T> {
|
||||
const outer = runningOperations.getStore() ?? [];
|
||||
if (outer.some((candidate) => candidate.cache === this && candidate.running)) {
|
||||
throw new Error(
|
||||
"CultCache refuses a re-entrant call: an attach, pull, registration, write or delete was started from " +
|
||||
"inside a store call, decoder, accessor or updater of an operation on the same cache, and would wait on itself.",
|
||||
);
|
||||
}
|
||||
|
||||
const current: RunningOperation = { cache: this, running: false };
|
||||
let advance!: () => void;
|
||||
const previous = this.#tail;
|
||||
this.#tail = new Promise<void>((resolve) => {
|
||||
advance = resolve;
|
||||
});
|
||||
return previous.then(() =>
|
||||
runningOperations.run([...outer, current], async () => {
|
||||
current.running = true;
|
||||
try {
|
||||
return await operation();
|
||||
} finally {
|
||||
current.running = false;
|
||||
advance();
|
||||
}
|
||||
}),
|
||||
);
|
||||
return previous.then(async () => {
|
||||
try {
|
||||
return await operation();
|
||||
} finally {
|
||||
advance();
|
||||
}
|
||||
});
|
||||
}
|
||||
|
||||
registerDocumentType<TDefinition extends AnyCultCacheDocumentDefinition>(
|
||||
|
|
|
|||
|
|
@ -1,5 +1,6 @@
|
|||
import assert from "node:assert/strict";
|
||||
import { exec, execFile } from "node:child_process";
|
||||
import { EventEmitter } from "node:events";
|
||||
import { existsSync } from "node:fs";
|
||||
import { access, mkdtemp, readFile, writeFile } from "node:fs/promises";
|
||||
import { homedir, tmpdir } from "node:os";
|
||||
|
|
@ -866,37 +867,44 @@ test("unawaitedFailingPutIsAnUnhandledRejection", async () => {
|
|||
await assert.rejects(cache.put(itemDocument, "k", { n: "y" }), /disk full/u);
|
||||
});
|
||||
|
||||
test("storePushThatAwaitsItsOwnCacheIsRefusedInsteadOfHanging", async () => {
|
||||
test("storeChangeHandlerWritingBackQueuesAfterTheOperation", async () => {
|
||||
const itemDocument = defineDocumentType({ type: "item", schema: z.object({ n: z.string() }) });
|
||||
const storePath = join(await mkdtemp(join(tmpdir(), "cultcache-reentrant-")), "generic.cc");
|
||||
const storePath = join(await mkdtemp(join(tmpdir(), "cultcache-writeback-")), "generic.cc");
|
||||
const inner = new SingleFileMessagePackBackingStore(storePath);
|
||||
let cache!: CultCache;
|
||||
const changes = new EventEmitter();
|
||||
const pushed: string[] = [];
|
||||
const store: CacheBackingStore = {
|
||||
pullAll: () => inner.pullAll(),
|
||||
delete: (entry) => inner.delete(entry),
|
||||
push: async (entry) => {
|
||||
await cache.put(itemDocument, "side", { n: "side" });
|
||||
await inner.push(entry);
|
||||
pushed.push(entry.key);
|
||||
changes.emit("pushed", entry.key);
|
||||
},
|
||||
};
|
||||
cache = CultCache.builder()
|
||||
const cache = CultCache.builder()
|
||||
.withRegistry(defineDocumentRegistry(itemDocument))
|
||||
.withGenericStore(store)
|
||||
.build();
|
||||
// The handler writes back and the store does not wait for it: that write queues behind the put.
|
||||
const writeBacks: Promise<unknown>[] = [];
|
||||
changes.on("pushed", (key: string) => {
|
||||
if (key === "k") {
|
||||
writeBacks.push(cache.put(itemDocument, "side", { n: "side" }));
|
||||
}
|
||||
});
|
||||
|
||||
const outcome = await Promise.race([
|
||||
cache.put(itemDocument, "k", { n: "x" }).then(() => "resolved", (error: unknown) => error),
|
||||
new Promise((resolve) => setTimeout(() => resolve("timed out"), 2000)),
|
||||
]);
|
||||
await cache.put(itemDocument, "k", { n: "x" });
|
||||
await Promise.all(writeBacks);
|
||||
|
||||
assert.ok(outcome instanceof Error, `expected a refusal, got ${String(outcome)}`);
|
||||
assert.match(outcome.message, /re-entrant call/u);
|
||||
assert.deepEqual(await inner.pullAll(), []);
|
||||
assert.deepEqual(cache.snapshot(), []);
|
||||
// Concurrent callers outside the operation are still queued, not refused.
|
||||
const plain = CultCache.builder().withRegistry(defineDocumentRegistry(itemDocument)).build();
|
||||
await Promise.all([plain.put(itemDocument, "a", { n: "a" }), plain.put(itemDocument, "b", { n: "b" })]);
|
||||
assert.equal(plain.getAll(itemDocument).length, 2);
|
||||
assert.equal(writeBacks.length, 1);
|
||||
assert.deepEqual(pushed, ["k", "side"]);
|
||||
assert.deepEqual(cache.get(itemDocument, "k"), { n: "x" });
|
||||
assert.deepEqual(cache.get(itemDocument, "side"), { n: "side" });
|
||||
assert.deepEqual(
|
||||
(await inner.pullAll()).map((entry) => entry.key).sort(),
|
||||
cache.snapshot().map((entry) => entry.key).sort(),
|
||||
);
|
||||
});
|
||||
|
||||
test("legacyKeyGlobalLoadsWithoutWritingAndFirstWriteLeavesOneGlobal", async () => {
|
||||
|
|
|
|||
|
|
@ -102,6 +102,13 @@ which route wrote it.
|
|||
- Direct calls on an attached store (`Push`, `Delete`, `PushAll`, `CommitBatch`,
|
||||
`PullAll`) take the same gate, so they wait for that cache's pull, flush or
|
||||
commit, and the cache's flush on dispose runs under it.
|
||||
- In every runtime, a store call must not wait for a write to its own cache to
|
||||
finish: that write queues behind the operation that is waiting for it, a
|
||||
cycle. A handler or subscriber that writes back without the store waiting for
|
||||
it is fine; its write queues and lands after the operation. Python refuses a
|
||||
mutation started on a thread already inside one of that cache's mutations
|
||||
with an error; no runtime detects a wait that escapes to another thread or
|
||||
async context.
|
||||
|
||||
## Dirtiness: loading never writes
|
||||
|
||||
|
|
@ -241,11 +248,11 @@ no serialization; a CultMath encoding, if added, uses this shape.
|
|||
| Zero stores is in memory | implemented | implemented | implemented | implemented |
|
||||
| Home cannot move after admission | implemented | implemented (`add_backing_store` returns `Err` and attaches nothing) | implemented (`addBackingStore` rejects and attaches nothing; it runs in the cache's serial queue, so it cannot move a write that is in flight) | implemented (`add_backing_store`/`add_generic_store` raise and attach nothing; they hold the cache's lock, so they cannot move a write that is in flight) |
|
||||
| Refused at load when not home | implemented | implemented (`pull_all_backing_stores` returns `Err` naming key and both stores; `load_soa` returns `Err` for a type with no home once a store is attached; both admit nothing and leave the prior view intact) | implemented (`pullAllBackingStores` throws naming key and both stores; admits nothing) | implemented (`pull_all_backing_stores` raises naming key and both stores; admits nothing) |
|
||||
| Refused write changes neither store nor cache | implemented | implemented (`put`, `put_envelope`, `put_raw_envelope`, `put_prepared_batch` and `delete` validate before the store call and apply infallibly after it; there are no user accessors) | implemented (home, `__global__` key and name and index accessors run before the store call; overwrite and delete use lookup values stored at admission; attach, pull, registration, writes and deletes run through one per-cache promise queue; one of those started from inside a store call, decoder, accessor or updater of the same cache's operation throws instead of waiting on itself, so stores must not await their cache) | implemented (same validation order and stored lookup keys as TypeScript; attach, pull, registration, writes, deletes, name and index resolution and snapshots hold one lock per cache; a mutation started on a thread already inside one of that cache's mutations raises, reads stay reentrant) |
|
||||
| Refused write changes neither store nor cache | implemented | implemented (`put`, `put_envelope`, `put_raw_envelope`, `put_prepared_batch` and `delete` validate before the store call and apply infallibly after it; there are no user accessors) | implemented (home, `__global__` key and name and index accessors run before the store call; overwrite and delete use lookup values stored at admission; attach, pull, registration, writes and deletes run through one per-cache promise queue; one of those started from inside an operation queues behind it, so a store that waits for it never finishes (see Locking)) | implemented (same validation order and stored lookup keys as TypeScript; attach, pull, registration, writes, deletes, name and index resolution and snapshots hold one lock per cache; a mutation started on a thread already inside one of that cache's mutations raises, reads stay reentrant; a wait on another thread is not detected (see Locking)) |
|
||||
| Attach hydrates; loading never writes | implemented | attach does not read (call `pull_all_backing_stores`); pull loads without writing | attach does not read (call `pullAllBackingStores`); pull loads without writing | attach does not read (call `pull_all_backing_stores`); pull loads without writing |
|
||||
| Runtime type decides schema | implemented | n/a (explicit type ids) | n/a | n/a |
|
||||
| Assignable lookups and watches | implemented | exact type ids | exact type ids | exact type ids |
|
||||
| Globals never invented, singleton | implemented | no globals | single `__global__`; a write under any other key is refused before any store is touched; replacing writes `__global__` in place; a second persisted global (any keys) is refused on pull; invents nothing; load compatibility: a pull adopts a global stored under another key as `__global__` without writing, and the first write or delete of it removes that record in the same call; removed once no store holds a non-`__global__` global | single `__global__`; `put`, `put_envelope` and `put_envelopes` refuse any other key before any store is touched; a second persisted global (any keys) is refused on pull; invents nothing; load compatibility: a pull adopts a global stored under another key as `__global__` without writing, and the first write or delete of it removes that record in the same call; removed once no store holds a non-`__global__` global |
|
||||
| Globals never invented, singleton | implemented | no globals | single `__global__`; a write under any other key is refused before any store is touched; replacing writes `__global__` in place; a second persisted global (any keys) is refused on pull; invents nothing; load compatibility: a pull adopts a global stored under another key as `__global__` without writing, and the first write of it pushes `__global__`, then deletes the legacy record, deleting `__global__` again if that delete fails (a delete of it removes only the legacy record). A crash between the two steps, or a failed compensation, leaves both records on disk, and the next pull refuses them loudly as a second global. Removed once no store holds a non-`__global__` global | single `__global__`; `put`, `put_envelope` and `put_envelopes` refuse any other key before any store is touched; a second persisted global (any keys) is refused on pull; invents nothing; load compatibility: a pull adopts a global stored under another key as `__global__` without writing, and the first write of it pushes `__global__`, then deletes the legacy record, deleting `__global__` again if that delete fails (a delete of it removes only the legacy record). A crash between the two steps, or a failed compensation, leaves both records on disk, and the next pull refuses them loudly as a second global. Removed once no store holds a non-`__global__` global |
|
||||
| Explicit batch commit | implemented | `put_prepared_batch`: staged records must share one home; that store receives one `push_all` of its own held records plus the batch; empty key or `stored_at` refused first; all-or-nothing on stores whose `push_all` is atomic, which the trait requires of every store: `push_all` has no default, and the bundled single-file and redb stores replace in one step | not implemented | `put_envelopes`, one type per call; every record validated before one `push_all`, applied only after the store accepts; atomicity on disk is the store's `push_all` |
|
||||
| Conditional commit | implemented | store-level `compare_exchange` family | not implemented | not implemented |
|
||||
| Per-assembly options, value-type arrays | contract | n/a | n/a | n/a |
|
||||
|
|
|
|||
Loading…
Reference in a new issue