cultcache-ts: update-with-retry (update-with-retry-ts) #169
16 changed files with 1342 additions and 106 deletions
|
|
@ -17,7 +17,7 @@ this page.
|
|||
| | CultCache | CultMath | Geometry | CultNet | CultMesh |
|
||||
| --- | --- | --- | --- | --- | --- |
|
||||
| **C#** | [first call, tested](#c-first-call); [update, tested](#c-update-with-retry) | [package](../../packages/cultmath/README.md); snippet not yet | [package](../../src/GameCult.Geometry/README.md); snippet not yet | [first call, tested](#c-cultnet-first-call); [package](../../src/GameCult.Networking) | [quickstart](../../src/GameCult.Mesh/docs/durable-node-quickstart.md); run by `scripts/verify-eve-getting-started.ps1` |
|
||||
| **TypeScript** | [first call, tested](#typescript-first-call) | not yet | not yet | [first call, tested](#typescript-cultnet-first-call); [package](../../packages/cultnet-ts/README.md) | [package](../../packages/cultmesh-ts/README.md); snippet not yet; [recipe: two-client .io game](io-game.md), tested with two live clients, browser page not yet run |
|
||||
| **TypeScript** | [first call, tested](#typescript-first-call); [update, tested](#typescript-update-with-retry) | not yet | not yet | [first call, tested](#typescript-cultnet-first-call); [package](../../packages/cultnet-ts/README.md) | [package](../../packages/cultmesh-ts/README.md); snippet not yet; [recipe: two-client .io game](io-game.md), tested with two live clients, browser page not yet run |
|
||||
| **Rust** | [first call, tested](#rust-first-call) | [native core](../../packages/cultmath/native/cultmath-core); snippet not yet | not yet | [first call, tested](#rust-cultnet-first-call); [package](../../packages/cultnet-rs/README.md), [examples](../../packages/cultnet-rs/examples) | [package](../../packages/cultmesh-rs); snippet not yet |
|
||||
| **Python** | [first call, tested](#python-first-call) | not yet | not yet | [first call, tested](#python-cultnet-first-call); [package](../../packages/cultnet-py/README.md) | [package](../../packages/cultmesh-py/README.md); snippet not yet |
|
||||
|
||||
|
|
@ -150,6 +150,30 @@ async function firstCall(path: string) {
|
|||
The test imports from `../src/index` where the block reads `@gamecult/cultcache-ts`; a second test in that file fails if the block stops matching the rest. More:
|
||||
`packages/cultcache-ts/README.md`.
|
||||
|
||||
## TypeScript update with retry
|
||||
|
||||
`update` changes a record from what the store holds now and commits only if it is still that record. The source is the
|
||||
second region of
|
||||
[`packages/cultcache-ts/test/first-call.test.ts`](../../packages/cultcache-ts/test/first-call.test.ts), which uses the
|
||||
definitions of the first call above; a test in that file fails if the block stops matching the file.
|
||||
|
||||
```ts
|
||||
async function restock(path: string) {
|
||||
const cache = CultCache.builder()
|
||||
.withRegistry(registry)
|
||||
.withGenericStore(new SingleFileMessagePackBackingStore(path))
|
||||
.build();
|
||||
await cache.pullAllBackingStores();
|
||||
return cache.update(itemDocument, "item:potion", (item) => {
|
||||
const current = item ?? { name: "Potion", category: "Consumable", value: 0 };
|
||||
return { ...current, value: current.value + 10 };
|
||||
});
|
||||
}
|
||||
```
|
||||
|
||||
The write is durable when the promise resolves. A lost race re-runs the function, so keep it free of side effects; when the
|
||||
attempts run out the result says `lostRace` instead of throwing. `put` is the blind opt-out: last writer wins.
|
||||
|
||||
## Rust first call
|
||||
|
||||
Same shape with `derive(DatabaseEntry)`: slots are `#[cultcache(key = N)]`, so the `.cc` payload is the
|
||||
|
|
|
|||
|
|
@ -5,13 +5,20 @@ All notable changes to `@gamecult/cultcache-ts` are recorded here, following
|
|||
|
||||
## [Unreleased]
|
||||
|
||||
### Added
|
||||
|
||||
- `CacheBackingStore` gains the optional `read(key)` (the record at the key with its content version, from one read) and `commitIf(entry, expectedVersion)` (write only if the version still holds, under the store's write lock; returns the storedAt written). The single-file store implements both. The content version is the lowercase hex SHA-256 of the record's persisted encoding `[key, schemaId, storedAt, payload]`, equal to the C# reference's on `tests/vectors/content-version`. A store without them still works for everything except `update`.
|
||||
- `CultCacheUpdateResult` and `CultCacheUpdateOptions` (`maxAttempts`, default 5), exported from the package entry and the browser entry.
|
||||
|
||||
### Changed
|
||||
|
||||
- A write that replaces a record at its key with different content (another schema id or other payload bytes) is stored strictly later than it: a supplied `storedAt` not later than the stored one is written one millisecond after it, so a content version that includes `storedAt` never repeats across writes. A rewrite of identical content (same schema id and payload bytes) keeps the stored stamp and so its version, which is what keeps a periodic flush of an unchanged cache from changing every record's version; C# re-mints every replacing write, so this is a departure from it, and the writer's type and catalog entry are written either way. `push`, `pushAll` and `commitIf` alike; the single-file store's `push` returns the stamp it wrote (`CacheBackingStore.push` may resolve `{ storedAt }`), and `put`, `putGlobal` and `putEnvelope` install that stamp in the cache, so the cache and the file agree after a put over a record stamped at or after now (a push result that is not a parseable stamp is ignored). A stamp that would pass 9999-12-31T23:59:59.999Z is refused with `RangeError` and nothing is written.
|
||||
- `withStoreWriteLock` looks for a link at the store path again once the lock is held, so a link put at the path while a writer waited on the lock is refused with `StoreLinkedError` and the lock released, instead of being replaced. Pinned by a test that puts the link while another holder owns the lock.
|
||||
- `CultCache.isWritable` does not look at store paths: a link at a store path is not predicted, and the write refuses it with `StoreLinkedError`. Pinned by a test through a cache attached to a linked store.
|
||||
|
||||
### Breaking
|
||||
|
||||
- `update` and `updateGlobal` are the conditional golden path and no longer the blind pair. `update(definition, key, apply, options?)` reads the record and its content version from the store, passes `apply` a fresh copy (never the held instance), commits only if the version still holds and runs `apply` again on a lost race, up to `maxAttempts`. `apply` may be async, may return `undefined` to write nothing, and must be free of side effects. It resolves to `{ outcome: "updated" | "declined" | "lostRace", document }` instead of the written value, so a caller reads `result.document`; a lost race that outlasts the attempts is `lostRace`, not a throw. `apply` returning `undefined` or `null` declines. `maxAttempts` is an integer of at least 1; `undefined` is the default 5 and `null` is refused with `RangeError`. A conditional write over a global loaded from a legacy key is refused wherever it enters (`updateGlobal`, or `update` at `CultCache.GLOBAL_KEY`) and leaves the legacy record untouched; `apply` may have run once before the write is refused. A store attached to a store-less cache while `apply` awaits makes that attempt a lost race. It needs a store with `read` and `commitIf` (the single-file store has them); a cache with no store compares the held entry. `put` and `putGlobal` are the blind writes and are unchanged.
|
||||
- A write at a single-file store path that is a symbolic link, live or dangling, is refused with the new `StoreLinkedError` (exported from the package entry; `error.path` names the path) before the lock and the replace, and writes nothing: `push`, `pushAll`, `delete` and a soft flush alike. Before, a write at a live link replaced the link with a regular file, a dangling link failed with `ENOENT` (or, for `pushAll`, was replaced), and a soft flush at a link skipped. A caller catching `ENOENT` around a write at a dangling link now catches `StoreLinkedError`; configure the link's target as the store path. Reads through a live link work. An `lstat` failure other than `ENOENT` (`ENOTDIR`, `EACCES`, `ELOOP`) is rethrown as it is.
|
||||
- `SingleFileMessagePackBackingStore.pushAll` writes the given entries onto the store as it is under the lock, as `push` does: a record under the same key is replaced and no other is removed. It used to replace the whole file with the entries, so `CultMesh.flush()` from a cache that held less than the store (a node created with `pullOnStart: false`, or one whose image was stale) wiped the records it never read. A caller that relied on `pushAll` to reset a store removes the file first. A soft flush is unchanged. `CultCache.storedSnapshot()` is the image as the stores hold it (a global loaded from a legacy key under that key) and is what `CultMesh.flush` persists, so a flush no longer leaves such a global under two keys; `snapshot()` is unchanged.
|
||||
- A single-file store marked `cultcache.store.v3` (it holds an element id, written by the C# runtime) is read, and a rewrite keeps the mark and the ids; this runtime does not decide when a document carries ids. A runtime from before this change refuses a v3 store by name, so rolling back to an older release after a store has been written as v3 is one-way: there is no carry-back, so restore from a copy taken before upgrading.
|
||||
|
|
@ -22,7 +29,7 @@ All notable changes to `@gamecult/cultcache-ts` are recorded here, following
|
|||
the many-row selection aliases (never indexed for lookup, same catalog `indexAlias`). The per-key migration rule is
|
||||
in the README; classify every key before the CultLib your tools load is updated. The refusal verdict is the contract, its text is informational.
|
||||
- The single-file reader (v1 and the legacy envelope array) refuses a store that holds two records under one key as `StoreUnreadableError`; a store that opened before may now be refused. Repair it by removing the extra record with the runtime that wrote it.
|
||||
- A `put`, `push` or `pushAll` that would land a second type under a held key (against the file, or within one batch) is refused with `SchemaConflictError`, and nothing is written. The written type may replace the record when its catalog entry carries the stored schema id as its own or lists it compatible (a type renamed under a stable id restamps the record, as C# does); otherwise retyping a key is a delete, then a put. The C# allowance for an undeclared older id of the same schema name that reads whole is not ported: a same-named type keeps passing, including one that drops persisted slots, which C# refuses.
|
||||
- A `put`, `push` or `pushAll` that would land a second type under a held key (against the file, or within one batch) is refused with `SchemaConflictError`, and nothing is written. The written type may replace the record when its catalog entry carries the stored schema id as its own or lists it compatible (a type renamed under a stable id restamps the record, as C# does, when its payload changes; an identical payload keeps the stamp); otherwise retyping a key is a delete, then a put. The C# allowance for an undeclared older id of the same schema name that reads whole is not ported: a same-named type keeps passing, including one that drops persisted slots, which C# refuses.
|
||||
- Same-type writes replace by key; in one `pushAll` batch the last entry for a key wins.
|
||||
- A store whose canonical-schema catalog slot is not a string is refused as `StoreUnreadableError`.
|
||||
- The cache holds one record per key (variants:ruling:cache-refuses-retype). A `put` or `putEnvelope` that lands a second type under a key the cache holds, or under a key a store carries foreign, is refused with `SchemaConflictError` naming the key, even with no store (a memory-only cache used to hold both types at the key). A type that owns the held record's schema id (as for the store) replaces it, unless the held record is in a store other than the written type's home: that rename is refused with `SchemaConflictError` (remove the record first). Two stores holding one key, registered or foreign, refuse the load and admit nothing. A store's `delete` removes the record whose stored schema id the removal's type owns, whatever the arrived record's `type` string.
|
||||
|
|
|
|||
|
|
@ -168,10 +168,8 @@ const settings = cache.getRequiredGlobal(settingsDocument);
|
|||
- `getByIndex(...)`
|
||||
- `getGlobal(...)`
|
||||
- `getRequiredGlobal(...)`
|
||||
- `put(...)`
|
||||
- `putGlobal(...)`
|
||||
- `update(...)`
|
||||
- `updateGlobal(...)`
|
||||
- `update(...)` and `updateGlobal(...)`, the conditional golden path: apply a function to a fresh copy of the record as the store holds it, commit only if it is unchanged, re-run on a lost race (`updated`, `declined` or `lostRace`)
|
||||
- `put(...)` and `putGlobal(...)`, the blind opt-out: last writer wins
|
||||
- `delete(...)`
|
||||
- `deleteGlobal(...)`
|
||||
- `snapshot()` (the logical image) and `storedSnapshot()` (the image as the stores hold it, for a whole flush)
|
||||
|
|
|
|||
|
|
@ -35,6 +35,8 @@ export type {
|
|||
InferCultCacheSchemaValue,
|
||||
CultCacheStoreRegistration,
|
||||
CultCacheUnreadSlots,
|
||||
CultCacheUpdateOptions,
|
||||
CultCacheUpdateResult,
|
||||
PushAllOptions,
|
||||
} from "./types";
|
||||
export type {
|
||||
|
|
|
|||
|
|
@ -22,6 +22,8 @@ import type {
|
|||
CultCacheSchemaCatalogEntry,
|
||||
CultCacheStoreRegistration,
|
||||
CultCacheUnreadSlots,
|
||||
CultCacheUpdateOptions,
|
||||
CultCacheUpdateResult,
|
||||
} from "./types";
|
||||
|
||||
type Accessor = (value: unknown) => string | undefined;
|
||||
|
|
@ -48,6 +50,10 @@ type EntryLookups = {
|
|||
readonly indexes: ReadonlyMap<string, string>;
|
||||
};
|
||||
|
||||
// What a conditional write requires of the record at its key: the content version its store reported (null: no record), or for a
|
||||
// cache with no store the entry it held. Every write installs a new entry object, so the identity catches an A-B-A.
|
||||
type WriteCondition = { readonly version: string | null } | { readonly held: HydratedEntry | undefined };
|
||||
|
||||
type HydratedEntry = CultCacheEnvelope & {
|
||||
catalogEntry: NonNullable<CultCacheEnvelope["catalogEntry"]>;
|
||||
value: unknown;
|
||||
|
|
@ -302,85 +308,88 @@ export class CultCache {
|
|||
// HydratedState, running every check and every user accessor, and replaces the cache's state
|
||||
// in one assignment only once the build finishes. A refused load admits nothing.
|
||||
pullAllBackingStores(): Promise<void> {
|
||||
return this.#serial(async () => {
|
||||
const next = emptyHydratedState();
|
||||
const persistedGlobalKeys = new Map<string, string>();
|
||||
return this.#serial(() => this.#pullNow());
|
||||
}
|
||||
|
||||
for (const registration of this.#stores) {
|
||||
const entries = await registration.store.pullAll();
|
||||
// The pull itself, for a caller already inside a serial operation.
|
||||
async #pullNow(): Promise<void> {
|
||||
const next = emptyHydratedState();
|
||||
const persistedGlobalKeys = new Map<string, string>();
|
||||
|
||||
for (const entry of entries) {
|
||||
const registered = this.#resolveDefinitionForEnvelope(entry);
|
||||
if (!registered) {
|
||||
// A store may hold what this cache has no type for; it is carried, not served, and not a reason to refuse the load.
|
||||
next.foreign.push({
|
||||
key: entry.key,
|
||||
schemaId: schemaIdForEnvelope(entry),
|
||||
schemaName: entry.catalogEntry?.schemaName ?? entry.type,
|
||||
storedAt: entry.storedAt,
|
||||
});
|
||||
continue;
|
||||
}
|
||||
for (const registration of this.#stores) {
|
||||
const entries = await registration.store.pullAll();
|
||||
|
||||
const type = registered.definition.type;
|
||||
const home = this.#resolveRegistration(type);
|
||||
if (home?.store !== registration.store) {
|
||||
for (const entry of entries) {
|
||||
const registered = this.#resolveDefinitionForEnvelope(entry);
|
||||
if (!registered) {
|
||||
// A store may hold what this cache has no type for; it is carried, not served, and not a reason to refuse the load.
|
||||
next.foreign.push({
|
||||
key: entry.key,
|
||||
schemaId: schemaIdForEnvelope(entry),
|
||||
schemaName: entry.catalogEntry?.schemaName ?? entry.type,
|
||||
storedAt: entry.storedAt,
|
||||
});
|
||||
continue;
|
||||
}
|
||||
|
||||
const type = registered.definition.type;
|
||||
const home = this.#resolveRegistration(type);
|
||||
if (home?.store !== registration.store) {
|
||||
throw new Error(
|
||||
`CultCache "${type}" record "${entry.key}" was loaded from ${this.#describe(registration)}, ` +
|
||||
`but its home is ${this.#describe(home)}.`,
|
||||
);
|
||||
}
|
||||
|
||||
const payload = this.#cloneBytes(entry.payload);
|
||||
const { value, unreadSlots } = registered.formatter.decode(payload);
|
||||
let admitted: CultCacheEnvelope = { ...entry, type, payload };
|
||||
if (registered.global) {
|
||||
const persistedKey = persistedGlobalKeys.get(type);
|
||||
if (persistedKey !== undefined) {
|
||||
throw new Error(
|
||||
`CultCache "${type}" record "${entry.key}" was loaded from ${this.#describe(registration)}, ` +
|
||||
`but its home is ${this.#describe(home)}.`,
|
||||
`CultCache global document type "${type}" has multiple persisted entries: "${persistedKey}" and "${entry.key}".`,
|
||||
);
|
||||
}
|
||||
|
||||
const payload = this.#cloneBytes(entry.payload);
|
||||
const { value, unreadSlots } = registered.formatter.decode(payload);
|
||||
let admitted: CultCacheEnvelope = { ...entry, type, payload };
|
||||
if (registered.global) {
|
||||
const persistedKey = persistedGlobalKeys.get(type);
|
||||
if (persistedKey !== undefined) {
|
||||
throw new Error(
|
||||
`CultCache global document type "${type}" has multiple persisted entries: "${persistedKey}" and "${entry.key}".`,
|
||||
);
|
||||
}
|
||||
persistedGlobalKeys.set(type, entry.key);
|
||||
if (entry.key !== CultCache.GLOBAL_KEY) {
|
||||
next.legacyGlobals.set(type, admitted);
|
||||
admitted = { ...admitted, key: CultCache.GLOBAL_KEY };
|
||||
}
|
||||
persistedGlobalKeys.set(type, entry.key);
|
||||
if (entry.key !== CultCache.GLOBAL_KEY) {
|
||||
next.legacyGlobals.set(type, admitted);
|
||||
admitted = { ...admitted, key: CultCache.GLOBAL_KEY };
|
||||
}
|
||||
const hydrated = this.#admitEntry(next, registered, admitted, value, unreadSlots);
|
||||
const other = this.#heldUnderOtherType(next, type, hydrated.key);
|
||||
if (other !== undefined) {
|
||||
throw new Error(`Record ${hydrated.key} is held by two stores, as type "${other.type}" and as type "${type}".`);
|
||||
}
|
||||
this.#installEntry(next, registered, hydrated);
|
||||
}
|
||||
}
|
||||
|
||||
const foreignByKey = new Map<string, string>();
|
||||
for (const record of next.foreign) {
|
||||
const first = foreignByKey.get(record.key);
|
||||
if (first !== undefined) {
|
||||
throw new Error(`Record ${record.key} is held by two stores, as schema "${first}" and as schema "${record.schemaName}".`);
|
||||
}
|
||||
foreignByKey.set(record.key, record.schemaName);
|
||||
const other = [...this.#definitions.keys()].map((type) => next.entries.get(this.#entryId(type, record.key))).find((held) => held !== undefined);
|
||||
const hydrated = this.#admitEntry(next, registered, admitted, value, unreadSlots);
|
||||
const other = this.#heldUnderOtherType(next, type, hydrated.key);
|
||||
if (other !== undefined) {
|
||||
throw new Error(`Record ${record.key} is held by two stores, as type "${other.type}" and as schema "${record.schemaName}".`);
|
||||
throw new Error(`Record ${hydrated.key} is held by two stores, as type "${other.type}" and as type "${type}".`);
|
||||
}
|
||||
this.#installEntry(next, registered, hydrated);
|
||||
}
|
||||
}
|
||||
|
||||
const foreignByKey = new Map<string, string>();
|
||||
for (const record of next.foreign) {
|
||||
const first = foreignByKey.get(record.key);
|
||||
if (first !== undefined) {
|
||||
throw new Error(`Record ${record.key} is held by two stores, as schema "${first}" and as schema "${record.schemaName}".`);
|
||||
}
|
||||
foreignByKey.set(record.key, record.schemaName);
|
||||
const other = [...this.#definitions.keys()].map((type) => next.entries.get(this.#entryId(type, record.key))).find((held) => held !== undefined);
|
||||
if (other !== undefined) {
|
||||
throw new Error(`Record ${record.key} is held by two stores, as type "${other.type}" and as schema "${record.schemaName}".`);
|
||||
}
|
||||
}
|
||||
|
||||
// Judged whole, once: the refusal names the least violation of the state this load would leave, whatever the
|
||||
// order of the stores and of the entries in them.
|
||||
refuseLeastIndexSharing(function* () {
|
||||
for (const entry of next.entries.values()) {
|
||||
for (const [alias, value] of entry.lookups.indexes) {
|
||||
yield { type: entry.type, schemaName: entry.catalogEntry.schemaName, alias, value, key: entry.key };
|
||||
}
|
||||
}
|
||||
}());
|
||||
|
||||
// Judged whole, once: the refusal names the least violation of the state this load would leave, whatever the
|
||||
// order of the stores and of the entries in them.
|
||||
refuseLeastIndexSharing(function* () {
|
||||
for (const entry of next.entries.values()) {
|
||||
for (const [alias, value] of entry.lookups.indexes) {
|
||||
yield { type: entry.type, schemaName: entry.catalogEntry.schemaName, alias, value, key: entry.key };
|
||||
}
|
||||
}
|
||||
}());
|
||||
|
||||
this.#hydrated = next;
|
||||
});
|
||||
this.#hydrated = next;
|
||||
}
|
||||
|
||||
get<TDefinition extends AnyCultCacheDocumentDefinition>(
|
||||
|
|
@ -513,12 +522,16 @@ export class CultCache {
|
|||
return key ? this.get(definition, key) : undefined;
|
||||
}
|
||||
|
||||
/**
|
||||
* The blind write: the record at the key becomes `value`, whatever it held, and the last writer wins. `update` is the
|
||||
* conditional path, which changes a record from what the store holds now and re-runs on a lost race.
|
||||
*/
|
||||
put<TDefinition extends AnyCultCacheDocumentDefinition>(
|
||||
definition: TDefinition,
|
||||
key: string,
|
||||
value: CultCacheDocumentValue<TDefinition>,
|
||||
): Promise<CultCacheDocumentValue<TDefinition>> {
|
||||
return this.#serial(() => this.#putNow(definition, key, value));
|
||||
return this.#serial(async () => (await this.#putNow(definition, key, value)).value);
|
||||
}
|
||||
|
||||
putEnvelope<TDefinition extends AnyCultCacheDocumentDefinition>(
|
||||
|
|
@ -560,31 +573,68 @@ export class CultCache {
|
|||
definition: TDefinition,
|
||||
value: CultCacheDocumentValue<TDefinition>,
|
||||
): Promise<CultCacheDocumentValue<TDefinition>> {
|
||||
return this.#serial(() => {
|
||||
return this.#serial(async () => {
|
||||
this.#requireGlobal(definition);
|
||||
return this.#putNow(definition, CultCache.GLOBAL_KEY, value);
|
||||
return (await this.#putNow(definition, CultCache.GLOBAL_KEY, value)).value;
|
||||
});
|
||||
}
|
||||
|
||||
update<TDefinition extends AnyCultCacheDocumentDefinition>(
|
||||
/**
|
||||
* Changes the record at the key from what the store holds now. `apply` may run up to `maxAttempts` times (default 5): each
|
||||
* attempt gets a fresh copy of the record as the store holds it (undefined when absent), which it may change and return, and
|
||||
* the result is written only if the record is still that version; a lost race runs `apply` again on the newer record. It must
|
||||
* not write this cache or have side effects. Returning undefined writes nothing (`declined`). A lost race that outlasts
|
||||
* `maxAttempts` returns `lostRace` and does not throw; a refusal, or a throw from `apply`, rejects and is not retried. The
|
||||
* write is durable when the promise resolves. `document` is the cache's record after the outcome. `put` is the blind write.
|
||||
*/
|
||||
async update<TDefinition extends AnyCultCacheDocumentDefinition>(
|
||||
definition: TDefinition,
|
||||
key: string,
|
||||
updater: (
|
||||
apply: (
|
||||
current: CultCacheDocumentValue<TDefinition> | undefined,
|
||||
) => CultCacheDocumentValue<TDefinition>,
|
||||
): Promise<CultCacheDocumentValue<TDefinition>> {
|
||||
return this.#serial(() => this.#putNow(definition, key, updater(this.get(definition, key))));
|
||||
) => CultCacheDocumentValue<TDefinition> | undefined | Promise<CultCacheDocumentValue<TDefinition> | undefined>,
|
||||
options: CultCacheUpdateOptions = {},
|
||||
): Promise<CultCacheUpdateResult<CultCacheDocumentValue<TDefinition>>> {
|
||||
if (typeof apply !== "function") {
|
||||
throw new TypeError("CultCache update needs a function to apply.");
|
||||
}
|
||||
// undefined is the default; anything else must be an integer of at least 1, so null is refused.
|
||||
const maxAttempts = options.maxAttempts === undefined ? 5 : options.maxAttempts;
|
||||
if (!Number.isInteger(maxAttempts) || maxAttempts < 1) {
|
||||
throw new RangeError(`CultCache update maxAttempts must be an integer of at least 1, not ${String(maxAttempts)}.`);
|
||||
}
|
||||
|
||||
for (let attempt = 1; ; attempt++) {
|
||||
const { expected, copy } = await this.#serial(() => this.#readForUpdate(definition, key));
|
||||
// Outside the serial queue and the store's lock: apply may await, and may write this cache through other keys.
|
||||
const next = await apply(copy as CultCacheDocumentValue<TDefinition> | undefined);
|
||||
if (next === undefined || next === null) {
|
||||
return { outcome: "declined", document: this.get(definition, key) };
|
||||
}
|
||||
const written = await this.#serial(() => this.#putNow(definition, key, next, expected));
|
||||
if (written.outcome === "committed") {
|
||||
return { outcome: "updated", document: this.get(definition, key) as CultCacheDocumentValue<TDefinition> };
|
||||
}
|
||||
if (attempt === maxAttempts) {
|
||||
// Display only. A cache with no store has nothing to pull, and a pull would replace what it holds with nothing.
|
||||
if (this.#stores.length > 0) {
|
||||
await this.#serial(() => this.#pullNow());
|
||||
}
|
||||
return { outcome: "lostRace", document: this.get(definition, key) };
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
updateGlobal<TDefinition extends AnyCultCacheDocumentDefinition>(
|
||||
/** `update` of the type's global record. */
|
||||
async updateGlobal<TDefinition extends AnyCultCacheDocumentDefinition>(
|
||||
definition: TDefinition,
|
||||
updater: (
|
||||
apply: (
|
||||
current: CultCacheDocumentValue<TDefinition> | undefined,
|
||||
) => CultCacheDocumentValue<TDefinition>,
|
||||
): Promise<CultCacheDocumentValue<TDefinition>> {
|
||||
return this.#serial(() =>
|
||||
this.#putNow(definition, CultCache.GLOBAL_KEY, updater(this.getGlobal(definition))),
|
||||
);
|
||||
) => CultCacheDocumentValue<TDefinition> | undefined | Promise<CultCacheDocumentValue<TDefinition> | undefined>,
|
||||
options: CultCacheUpdateOptions = {},
|
||||
): Promise<CultCacheUpdateResult<CultCacheDocumentValue<TDefinition>>> {
|
||||
this.#requireGlobal(definition);
|
||||
return this.update(definition, CultCache.GLOBAL_KEY, apply, options);
|
||||
}
|
||||
|
||||
delete<TDefinition extends AnyCultCacheDocumentDefinition>(
|
||||
|
|
@ -724,11 +774,55 @@ export class CultCache {
|
|||
this.#stores.push(candidate);
|
||||
}
|
||||
|
||||
// One attempt's read for update: the record at the key as its home holds it now, the condition a commit must meet, and a copy of
|
||||
// the record for apply to change. A store-backed cache asks its store, which reports the record and its version from one read;
|
||||
// a cache with no store has only the held entry, whose identity is the condition and whose admitted payload is the copy.
|
||||
async #readForUpdate(
|
||||
definition: AnyCultCacheDocumentDefinition,
|
||||
key: string,
|
||||
): Promise<{ expected: WriteCondition; copy: unknown }> {
|
||||
const registered = this.#requireDefinition(definition);
|
||||
this.#requireGlobalKey(registered, key);
|
||||
const type = definition.type;
|
||||
this.#refuseLegacyConditional(type);
|
||||
const home = this.#homeStore(type);
|
||||
if (home === undefined) {
|
||||
const held = this.#hydrated.entries.get(this.#entryId(type, key));
|
||||
if (held === undefined && this.#heldUnderOtherType(this.#hydrated, type, key) !== undefined) {
|
||||
throw new Error(`Record ${key} is not a "${type}"; update cannot copy it.`);
|
||||
}
|
||||
return { expected: { held }, copy: held ? registered.formatter.decode(this.#cloneBytes(held.payload)).value : undefined };
|
||||
}
|
||||
if (home.read === undefined || home.commitIf === undefined) {
|
||||
throw new Error(`CultCache backing store for type "${type}" cannot write conditionally; update needs a store with read and commitIf.`);
|
||||
}
|
||||
const found = await home.read(key);
|
||||
if (found === undefined) {
|
||||
return { expected: { version: null }, copy: undefined };
|
||||
}
|
||||
if (this.#resolveDefinitionForEnvelope(found.envelope) !== registered) {
|
||||
throw new Error(`Record ${key} is not a "${type}"; update cannot copy it.`);
|
||||
}
|
||||
return { expected: { version: found.version }, copy: registered.formatter.decode(this.#cloneBytes(found.envelope.payload)).value };
|
||||
}
|
||||
|
||||
#putNow<TDefinition extends AnyCultCacheDocumentDefinition>(
|
||||
definition: TDefinition,
|
||||
key: string,
|
||||
value: CultCacheDocumentValue<TDefinition>,
|
||||
): Promise<{ outcome: "committed"; value: CultCacheDocumentValue<TDefinition> }>;
|
||||
#putNow<TDefinition extends AnyCultCacheDocumentDefinition>(
|
||||
definition: TDefinition,
|
||||
key: string,
|
||||
value: CultCacheDocumentValue<TDefinition>,
|
||||
expected: WriteCondition,
|
||||
): Promise<{ outcome: "committed"; value: CultCacheDocumentValue<TDefinition> } | { outcome: "mismatch" }>;
|
||||
async #putNow<TDefinition extends AnyCultCacheDocumentDefinition>(
|
||||
definition: TDefinition,
|
||||
key: string,
|
||||
value: CultCacheDocumentValue<TDefinition>,
|
||||
): Promise<CultCacheDocumentValue<TDefinition>> {
|
||||
expected?: WriteCondition,
|
||||
): Promise<{ outcome: "committed"; value: CultCacheDocumentValue<TDefinition> } | { outcome: "mismatch" }> {
|
||||
const registered = this.#requireDefinition(definition);
|
||||
const payload = registered.formatter.encode(value);
|
||||
const { value: parsed, unreadSlots } = registered.formatter.decode(payload) as { value: CultCacheDocumentValue<TDefinition>; unreadSlots?: NotedSlots };
|
||||
|
|
@ -740,19 +834,55 @@ export class CultCache {
|
|||
schemaId: registered.catalogEntry.schemaId,
|
||||
catalogEntry: registered.catalogEntry,
|
||||
};
|
||||
await this.#writeNow(registered, entry, parsed, unreadSlots);
|
||||
return parsed;
|
||||
const outcome = await this.#writeNow(registered, entry, parsed, unreadSlots, expected);
|
||||
return outcome === "committed" ? { outcome, value: parsed } : { outcome };
|
||||
}
|
||||
|
||||
// Validates the whole write (home, global singleton, name and index accessors) before the store
|
||||
// is touched; once the store accepts, applying to the cache has no fallible step.
|
||||
async #writeNow(registered: RegisteredDefinition, entry: CultCacheEnvelope, value: unknown, unreadSlots?: NotedSlots): Promise<void> {
|
||||
// With a condition, the write lands only if the record at the key is still the one the caller read: the store's version
|
||||
// (commitIf), or for a cache with no store the held entry itself. Otherwise nothing is written or installed and the answer is
|
||||
// "mismatch". Without one the write is blind, as put's.
|
||||
async #writeNow(
|
||||
registered: RegisteredDefinition,
|
||||
entry: CultCacheEnvelope,
|
||||
value: unknown,
|
||||
unreadSlots?: NotedSlots,
|
||||
expected?: WriteCondition,
|
||||
): Promise<"committed" | "mismatch"> {
|
||||
const home = this.#homeStore(entry.type);
|
||||
const retyped = this.#judge(registered, entry.key, entry);
|
||||
const hydrated = this.#admitEntry(this.#hydrated, registered, entry, value, unreadSlots);
|
||||
this.#refuseIndexSharing(this.#hydrated, entry.type, registered.catalogEntry.schemaName, entry.key, hydrated.lookups);
|
||||
const legacy = this.#hydrated.legacyGlobals.get(entry.type);
|
||||
await home?.push(entry);
|
||||
// A conditional write over a legacy global is refused here too (a pull while apply awaited can have loaded one): the condition
|
||||
// was read from the global's own key, the record sits under its legacy key, and the retire below would delete it unseen.
|
||||
if (expected !== undefined) {
|
||||
this.#refuseLegacyConditional(entry.type);
|
||||
}
|
||||
if (expected !== undefined && "held" in expected) {
|
||||
// The condition was fixed at read for a cache with no store. A store attached since has its own record at the key that
|
||||
// this read never saw, so the held identity no longer judges the write: lose the race and read again.
|
||||
if (home !== undefined || this.#hydrated.entries.get(this.#entryId(entry.type, entry.key)) !== expected.held) {
|
||||
return "mismatch";
|
||||
}
|
||||
}
|
||||
if (expected !== undefined && "version" in expected) {
|
||||
if (home?.commitIf === undefined) {
|
||||
throw new Error(`CultCache backing store for type "${entry.type}" cannot write conditionally; update needs a store with read and commitIf.`);
|
||||
}
|
||||
const committed = await home.commitIf(entry, expected.version);
|
||||
if (committed.outcome === "mismatch") {
|
||||
return "mismatch";
|
||||
}
|
||||
hydrated.storedAt = committed.storedAt;
|
||||
} else {
|
||||
// A store that re-stamps a write (storedAt later than the record it replaces) reports the stamp; the cache holds that one.
|
||||
const pushed = await home?.push(entry);
|
||||
if (typeof pushed === "object" && pushed !== null && typeof pushed.storedAt === "string" && !Number.isNaN(Date.parse(pushed.storedAt))) {
|
||||
hydrated.storedAt = pushed.storedAt;
|
||||
}
|
||||
}
|
||||
if (home && legacy) {
|
||||
await this.#retireLegacyGlobal(home, legacy, entry);
|
||||
}
|
||||
|
|
@ -762,6 +892,17 @@ export class CultCache {
|
|||
this.#removeEntry(this.#hydrated, this.#definitions.get(retyped.type)!, retyped);
|
||||
}
|
||||
this.#installEntry(this.#hydrated, registered, hydrated);
|
||||
return "committed";
|
||||
}
|
||||
|
||||
// The one rule that a conditional write (update, whichever entry reaches it) never touches a global loaded from a legacy key:
|
||||
// only putGlobal retires that key. Asked before apply runs and again at the write.
|
||||
#refuseLegacyConditional(type: string): void {
|
||||
if (this.#hydrated.legacyGlobals.has(type)) {
|
||||
throw new Error(
|
||||
`CultCache global "${type}" was loaded from a legacy key; write it once with putGlobal to retire its legacy key.`,
|
||||
);
|
||||
}
|
||||
}
|
||||
|
||||
// An envelope of the definition's type at the key, carrying nothing: what the judge asks about a write or a removal by that type.
|
||||
|
|
|
|||
|
|
@ -1,3 +1,4 @@
|
|||
import { createHash } from "node:crypto";
|
||||
import { lstatSync, readFileSync } from "node:fs";
|
||||
import { mkdir } from "node:fs/promises";
|
||||
import { dirname, resolve } from "node:path";
|
||||
|
|
@ -55,14 +56,42 @@ export class SingleFileMessagePackBackingStore implements CacheBackingStore {
|
|||
return disk.envelopes;
|
||||
}
|
||||
|
||||
async push(entry: CultCacheEnvelope): Promise<void> {
|
||||
await this.#enqueue(async () => {
|
||||
async push(entry: CultCacheEnvelope): Promise<{ storedAt: string }> {
|
||||
return this.#enqueue(async () => {
|
||||
const disk = this.#readDisk();
|
||||
this.#format = disk.format;
|
||||
const existing = disk.envelopes;
|
||||
requireOneTypePerKey(this.filePath, [entry], existing);
|
||||
const filtered = existing.filter((candidate) => candidate.key !== entry.key);
|
||||
await this.#writeAll([entry], filtered);
|
||||
const [written] = await this.#writeAll([entry], existing);
|
||||
return { storedAt: written!.storedAt };
|
||||
});
|
||||
}
|
||||
|
||||
// The record at the key as the file holds it now, with its content version, from one read. Like pullAll it takes no lock: a
|
||||
// reader sees a whole file (the commit renames one over the other), and the version it reports is the one commitIf judges.
|
||||
async read(key: string): Promise<{ envelope: CultCacheEnvelope; version: string } | undefined> {
|
||||
const envelope = readSingleFileStore(this.filePath).envelopes.find((candidate) => candidate.key === key);
|
||||
return envelope === undefined ? undefined : { envelope, version: recordVersion(envelope) };
|
||||
}
|
||||
|
||||
// Writes `entry` the way push does, but only if the record at its key (of any type) still has `expectedVersion`; null expects
|
||||
// none. The comparison and the write are one hold of the write lock, so no other writer lands between them.
|
||||
async commitIf(
|
||||
entry: CultCacheEnvelope,
|
||||
expectedVersion: string | null,
|
||||
): Promise<{ outcome: "committed"; storedAt: string } | { outcome: "mismatch" }> {
|
||||
return this.#enqueue(async () => {
|
||||
const disk = this.#readDisk();
|
||||
this.#format = disk.format;
|
||||
const existing = disk.envelopes;
|
||||
const current = existing.find((candidate) => candidate.key === entry.key);
|
||||
if ((current === undefined ? null : recordVersion(current)) !== expectedVersion) {
|
||||
return { outcome: "mismatch" as const };
|
||||
}
|
||||
|
||||
requireOneTypePerKey(this.filePath, [entry], existing);
|
||||
const [written] = await this.#writeAll([entry], existing);
|
||||
return { outcome: "committed" as const, storedAt: written!.storedAt };
|
||||
});
|
||||
}
|
||||
|
||||
|
|
@ -126,13 +155,67 @@ export class SingleFileMessagePackBackingStore implements CacheBackingStore {
|
|||
return result;
|
||||
}
|
||||
|
||||
async #writeAll(supplied: CultCacheEnvelope[], arrived: CultCacheEnvelope[]): Promise<void> {
|
||||
// Writes the supplied records onto the arrived ones and returns the supplied records as written. A supplied record whose
|
||||
// schema id and payload bytes differ from the arrived one it replaces is never stamped earlier than the record it replaces, so a content version that includes storedAt
|
||||
// cannot repeat across writes (variants:ruling:storedat-on-change): one not later than the arrived stamp is written one
|
||||
// millisecond after it.
|
||||
async #writeAll(supplied: CultCacheEnvelope[], arrived: CultCacheEnvelope[]): Promise<CultCacheEnvelope[]> {
|
||||
const arrivedByKey = new Map(arrived.map((entry) => [entry.key, entry]));
|
||||
const written = supplied.map((entry) => {
|
||||
const prior = arrivedByKey.get(entry.key);
|
||||
if (prior === undefined) {
|
||||
return entry;
|
||||
}
|
||||
// A rewrite of the identical record (same schema id and payload bytes) keeps the stamp and so the version it has; the
|
||||
// writer's type and catalog entry still go down with it.
|
||||
if (sameContent(entry, prior)) {
|
||||
return { ...entry, storedAt: prior.storedAt };
|
||||
}
|
||||
return { ...entry, storedAt: laterStoredAt(entry.storedAt, prior.storedAt) };
|
||||
});
|
||||
await mkdir(dirname(this.filePath), { recursive: true });
|
||||
|
||||
await commitStoreFile(this.filePath, encode(encodeSnapshot(supplied, arrived, this.#format)));
|
||||
await commitStoreFile(this.filePath, encode(encodeSnapshot(written, arrived, this.#format)));
|
||||
return written;
|
||||
}
|
||||
}
|
||||
|
||||
// The content version of a stored record: the lowercase hex SHA-256 of its persisted encoding, the MessagePack array
|
||||
// [key, schemaId, storedAt, payload] as the store writes it (the C# reference's StoredVersion). Private to this module so
|
||||
// node:crypto stays out of the browser-safe cache.
|
||||
export function recordVersion(entry: CultCacheEnvelope): string {
|
||||
return createHash("sha256")
|
||||
.update(encode([entry.key, schemaIdForEnvelope(entry), entry.storedAt, documentPayloadBytes(entry.payload)]))
|
||||
.digest("hex");
|
||||
}
|
||||
|
||||
function sameContent(a: CultCacheEnvelope, b: CultCacheEnvelope): boolean {
|
||||
if (schemaIdForEnvelope(a) !== schemaIdForEnvelope(b)) {
|
||||
return false;
|
||||
}
|
||||
const x = documentPayloadBytes(a.payload);
|
||||
const y = documentPayloadBytes(b.payload);
|
||||
return x.length === y.length && x.every((byte, i) => byte === y[i]);
|
||||
}
|
||||
|
||||
// `storedAt` if it is later than `prior`, else `prior` plus one millisecond. A stamp that does not parse is not later; a prior
|
||||
// that does not parse leaves nothing to be later than.
|
||||
function laterStoredAt(storedAt: string, prior: string): string {
|
||||
const priorMs = Date.parse(prior);
|
||||
if (Number.isNaN(priorMs)) {
|
||||
return storedAt;
|
||||
}
|
||||
const storedMs = Date.parse(storedAt);
|
||||
if (!(Number.isNaN(storedMs) || storedMs <= priorMs)) {
|
||||
return storedAt;
|
||||
}
|
||||
// The stamp is ISO-8601 with a four-digit year; one past 9999-12-31T23:59:59.999Z would print as an extended year.
|
||||
if (priorMs + 1 > Date.UTC(9999, 11, 31, 23, 59, 59, 999)) {
|
||||
throw new RangeError(`CultCache cannot stamp a write later than the record it replaces (${prior}): the stamp would leave the four-digit years.`);
|
||||
}
|
||||
return new Date(priorMs + 1).toISOString();
|
||||
}
|
||||
|
||||
// A store holds a key once, and the reader refuses a file that holds it twice, so a write that would put a second type under a
|
||||
// key (against a record on disk, or within its own batch) is refused before anything is written. Retyping a record is a delete
|
||||
// of the old type followed by a put of the new one.
|
||||
|
|
|
|||
|
|
@ -125,9 +125,34 @@ export interface PushAllOptions {
|
|||
|
||||
export interface CacheBackingStore {
|
||||
pullAll(): Promise<CultCacheEnvelope[]>;
|
||||
push(entry: CultCacheEnvelope): Promise<void>;
|
||||
/** A store that re-stamps the write returns the storedAt it wrote; the cache then holds that stamp, not its own. */
|
||||
push(entry: CultCacheEnvelope): Promise<void | { storedAt: string }>;
|
||||
delete(entry: CultCacheEnvelope): Promise<void>;
|
||||
pushAll?(entries: CultCacheEnvelope[], options?: PushAllOptions): Promise<void>;
|
||||
/**
|
||||
* The record at the key as the store holds it now, with its content version, from one read; undefined when the store holds
|
||||
* none. The version is the store's own: the cache carries it from `read` to `commitIf` and never computes it.
|
||||
*/
|
||||
read?(key: string): Promise<{ envelope: CultCacheEnvelope; version: string } | undefined>;
|
||||
/**
|
||||
* Writes the entry only if the record at its key still has `expectedVersion` (null: no record), judged and written under the
|
||||
* store's write lock. A mismatch writes nothing. A commit returns the storedAt actually written.
|
||||
*/
|
||||
commitIf?(
|
||||
entry: CultCacheEnvelope,
|
||||
expectedVersion: string | null,
|
||||
): Promise<{ outcome: "committed"; storedAt: string } | { outcome: "mismatch" }>;
|
||||
}
|
||||
|
||||
/** The outcome of `CultCache.update`: what `apply` wrote, declined to write, or lost to a rival writer. */
|
||||
export type CultCacheUpdateResult<TValue> =
|
||||
| { outcome: "updated"; document: TValue }
|
||||
| { outcome: "declined"; document: TValue | undefined }
|
||||
| { outcome: "lostRace"; document: TValue | undefined };
|
||||
|
||||
export interface CultCacheUpdateOptions {
|
||||
/** How many times `apply` may run before a lost race is returned as `lostRace`. An integer of at least 1; default 5. */
|
||||
maxAttempts?: number;
|
||||
}
|
||||
|
||||
export interface CultCacheStoreRegistration {
|
||||
|
|
|
|||
|
|
@ -46,6 +46,21 @@ async function firstCall(path: string) {
|
|||
}
|
||||
// snippet:end
|
||||
|
||||
// The update snippet, which docs/start/README.md shows after the first call and which uses its definitions.
|
||||
// snippet:update:begin
|
||||
async function restock(path: string) {
|
||||
const cache = CultCache.builder()
|
||||
.withRegistry(registry)
|
||||
.withGenericStore(new SingleFileMessagePackBackingStore(path))
|
||||
.build();
|
||||
await cache.pullAllBackingStores();
|
||||
return cache.update(itemDocument, "item:potion", (item) => {
|
||||
const current = item ?? { name: "Potion", category: "Consumable", value: 0 };
|
||||
return { ...current, value: current.value + 10 };
|
||||
});
|
||||
}
|
||||
// snippet:update:end
|
||||
|
||||
test("define, store and read back a typed document", async () => {
|
||||
const dir = await mkdtemp(join(tmpdir(), "cultcache-first-call-"));
|
||||
try {
|
||||
|
|
@ -57,13 +72,35 @@ test("define, store and read back a typed document", async () => {
|
|||
}
|
||||
});
|
||||
|
||||
test("docs show the tested snippet", async () => {
|
||||
test("update with retry: restock twice reads value 20 through a second cache", async () => {
|
||||
const dir = await mkdtemp(join(tmpdir(), "cultcache-first-call-"));
|
||||
try {
|
||||
const path = join(dir, "restock.cc");
|
||||
assert.equal((await restock(path)).outcome, "updated");
|
||||
assert.equal((await restock(path)).outcome, "updated");
|
||||
const reopened = CultCache.builder()
|
||||
.withRegistry(registry)
|
||||
.withGenericStore(new SingleFileMessagePackBackingStore(path))
|
||||
.build();
|
||||
await reopened.pullAllBackingStores();
|
||||
assert.equal(reopened.get(itemDocument, "item:potion")?.value, 20);
|
||||
} finally {
|
||||
await rm(dir, { recursive: true, force: true });
|
||||
}
|
||||
});
|
||||
|
||||
// The docs show the region between `// snippet:<region>begin` and `end` markers verbatim, the package import aside.
|
||||
async function assertDocsShow(region: "" | "update:") {
|
||||
// Compiled to dist-test/test/first-call.test.js, so the package root is two levels up.
|
||||
const source = await readFile(join(__dirname, "..", "..", "test", "first-call.test.ts"), "utf8");
|
||||
const snippet = source
|
||||
.split("// snippet:begin\n")[1]
|
||||
.split("// snippet:end\n")[0]
|
||||
.split(`// snippet:${region}begin\n`)[1]
|
||||
.split(`// snippet:${region}end\n`)[0]
|
||||
.replace('"../src/index"', '"@gamecult/cultcache-ts"');
|
||||
const docs = (await readFile(join(__dirname, "..", "..", "..", "..", "docs", "start", "README.md"), "utf8")).replace(/\r\n/g, "\n");
|
||||
assert.ok(docs.includes("```ts\n" + snippet + "```"), "docs/start/README.md no longer shows test/first-call.test.ts");
|
||||
});
|
||||
assert.ok(docs.includes("```ts\n" + snippet + "```"), `docs/start/README.md no longer shows the ${region || "first-call "}region of test/first-call.test.ts`);
|
||||
}
|
||||
|
||||
test("docs show the tested snippet", () => assertDocsShow(""));
|
||||
|
||||
test("docs show the tested update snippet", () => assertDocsShow("update:"));
|
||||
|
|
|
|||
175
packages/cultcache-ts/test/store-conditional.test.ts
Normal file
175
packages/cultcache-ts/test/store-conditional.test.ts
Normal file
|
|
@ -0,0 +1,175 @@
|
|||
import assert from "node:assert/strict";
|
||||
import { createHash } from "node:crypto";
|
||||
import { existsSync, readFileSync } from "node:fs";
|
||||
import { copyFile, mkdtemp, readFile, rm } from "node:fs/promises";
|
||||
import { tmpdir } from "node:os";
|
||||
import { join, resolve } from "node:path";
|
||||
import test from "node:test";
|
||||
|
||||
import { decode, encode } from "@msgpack/msgpack";
|
||||
|
||||
import { documentPayloadBytes, documentPayloadFromBytes } from "../src/document";
|
||||
import { SingleFileMessagePackBackingStore, readSingleFileStore, recordVersion } from "../src/single-file-messagepack-backing-store";
|
||||
import { schemaIdForEnvelope } from "../src/store-format";
|
||||
import type { CultCacheEnvelope } from "../src/types";
|
||||
|
||||
// The store owns the content version and the conditional verdict (variants:ruling:write-set-content-version-declared-ownership):
|
||||
// read reports a record with its version, commitIf writes only while the version holds, and a write that replaces a record
|
||||
// is stamped later than it (variants:ruling:storedat-on-change).
|
||||
|
||||
function findVectors(from: string): string {
|
||||
for (let dir = from; ; dir = resolve(dir, "..")) {
|
||||
if (existsSync(join(dir, "tests", "vectors", "content-version"))) return join(dir, "tests", "vectors", "content-version");
|
||||
if (resolve(dir, "..") === dir) throw new Error("tests/vectors/content-version not found above the test.");
|
||||
}
|
||||
}
|
||||
|
||||
const entry = (key: string, byte: number, storedAt = "2026-01-01T00:00:00.000Z", type = "t"): CultCacheEnvelope => ({
|
||||
key,
|
||||
type,
|
||||
payload: documentPayloadFromBytes(Uint8Array.of(0x91, byte)),
|
||||
storedAt,
|
||||
});
|
||||
|
||||
async function scratch(): Promise<string> {
|
||||
return mkdtemp(join(tmpdir(), "cultcache-conditional-"));
|
||||
}
|
||||
|
||||
test("the content version of every record of the C#-written store equals the version the reference read", async () => {
|
||||
const vectors = findVectors(__dirname);
|
||||
const rows = decode(readFileSync(join(vectors, "versions.msgpack"))) as Array<[string, string]>;
|
||||
assert.equal(rows.length, 2);
|
||||
const directory = await scratch();
|
||||
try {
|
||||
const path = join(directory, "store.cc");
|
||||
await copyFile(join(vectors, "store.cc"), path);
|
||||
const store = new SingleFileMessagePackBackingStore(path);
|
||||
for (const [key, version] of rows) {
|
||||
const found = await store.read(key);
|
||||
assert.equal(found?.version, version, `read(${key}).version`);
|
||||
assert.equal(recordVersion(found!.envelope), version, `recordVersion(${key})`);
|
||||
}
|
||||
assert.equal(await store.read("absent"), undefined);
|
||||
} finally {
|
||||
await rm(directory, { recursive: true, force: true });
|
||||
}
|
||||
});
|
||||
|
||||
test("the version is the SHA-256 of the record's persisted encoding and changes with every field of it", async () => {
|
||||
const base = entry("k", 1);
|
||||
const hex = (e: CultCacheEnvelope) =>
|
||||
createHash("sha256").update(encode([e.key, schemaIdForEnvelope(e), e.storedAt, documentPayloadBytes(e.payload)])).digest("hex");
|
||||
assert.equal(recordVersion(base), hex(base));
|
||||
const versions = new Set([base, entry("k", 2), entry("k", 1, "2026-01-01T00:00:00.001Z"), entry("j", 1), { ...base, schemaId: "sha256:other" }].map(recordVersion));
|
||||
assert.equal(versions.size, 5);
|
||||
});
|
||||
|
||||
test("commitIf writes nothing when the version does not hold, and commits when it does", async () => {
|
||||
const directory = await scratch();
|
||||
try {
|
||||
const path = join(directory, "store.cc");
|
||||
const store = new SingleFileMessagePackBackingStore(path);
|
||||
// null expects no record: it commits over an absent key and mismatches over a present one.
|
||||
assert.deepEqual(await store.commitIf(entry("k", 1), "not-a-version"), { outcome: "mismatch" });
|
||||
assert.equal(existsSync(path), false, "a mismatch creates no file");
|
||||
assert.equal((await store.commitIf(entry("k", 1), null)).outcome, "committed");
|
||||
const first = (await store.read("k"))!;
|
||||
const before = await readFile(path);
|
||||
assert.deepEqual(await store.commitIf(entry("k", 2), null), { outcome: "mismatch" });
|
||||
assert.deepEqual(await store.commitIf(entry("k", 2), "0".repeat(64)), { outcome: "mismatch" });
|
||||
assert.deepEqual([...(await readFile(path))], [...before], "a mismatch leaves the bytes identical");
|
||||
assert.equal((await store.commitIf(entry("k", 2), first.version)).outcome, "committed");
|
||||
const second = (await store.read("k"))!;
|
||||
assert.notEqual(second.version, first.version);
|
||||
assert.deepEqual([...documentPayloadBytes(second.envelope.payload)], [0x91, 2]);
|
||||
} finally {
|
||||
await rm(directory, { recursive: true, force: true });
|
||||
}
|
||||
});
|
||||
|
||||
test("commitIf judges the record at the key whatever its type, and a record of another key does not matter", async () => {
|
||||
const directory = await scratch();
|
||||
try {
|
||||
const store = new SingleFileMessagePackBackingStore(join(directory, "store.cc"));
|
||||
await store.push(entry("a", 1, undefined, "u"));
|
||||
await store.push(entry("b", 1));
|
||||
const other = (await store.read("a"))!;
|
||||
assert.deepEqual(await store.commitIf(entry("b", 2), other.version), { outcome: "mismatch" });
|
||||
assert.equal((await store.commitIf(entry("b", 2), recordVersion((await store.read("b"))!.envelope))).outcome, "committed");
|
||||
assert.deepEqual([...documentPayloadBytes((await store.read("a"))!.envelope.payload)], [0x91, 1], "the record at another key is untouched");
|
||||
} finally {
|
||||
await rm(directory, { recursive: true, force: true });
|
||||
}
|
||||
});
|
||||
|
||||
test("a write that replaces a record is stored strictly later than it, whatever the write's own storedAt", async () => {
|
||||
const directory = await scratch();
|
||||
try {
|
||||
const store = new SingleFileMessagePackBackingStore(join(directory, "store.cc"));
|
||||
const at = "2026-03-04T05:06:07.100Z";
|
||||
await store.push(entry("k", 1, at));
|
||||
// Not later (equal, earlier, unparseable) is written one millisecond after the record it replaces, by push, pushAll and commitIf.
|
||||
await store.push(entry("k", 2, at));
|
||||
assert.equal((await store.read("k"))!.envelope.storedAt, "2026-03-04T05:06:07.101Z");
|
||||
await store.push(entry("k", 3, "2020-01-01T00:00:00.000Z"));
|
||||
assert.equal((await store.read("k"))!.envelope.storedAt, "2026-03-04T05:06:07.102Z");
|
||||
await store.pushAll([entry("k", 4, "not a time")]);
|
||||
assert.equal((await store.read("k"))!.envelope.storedAt, "2026-03-04T05:06:07.103Z");
|
||||
const committed = await store.commitIf(entry("k", 5, "2020-01-01T00:00:00.000Z"), (await store.read("k"))!.version);
|
||||
assert.deepEqual(committed, { outcome: "committed", storedAt: "2026-03-04T05:06:07.104Z" });
|
||||
assert.equal((await store.read("k"))!.envelope.storedAt, "2026-03-04T05:06:07.104Z");
|
||||
// A later stamp is kept as it is, and a key with no record before it is stamped as it is.
|
||||
await store.push(entry("k", 6, "2030-01-01T00:00:00.000Z"));
|
||||
await store.push(entry("fresh", 1, "2000-01-01T00:00:00.000Z"));
|
||||
assert.equal((await store.read("k"))!.envelope.storedAt, "2030-01-01T00:00:00.000Z");
|
||||
assert.equal((await store.read("fresh"))!.envelope.storedAt, "2000-01-01T00:00:00.000Z");
|
||||
assert.equal(readSingleFileStore(join(directory, "store.cc")).envelopes.length, 2);
|
||||
} finally {
|
||||
await rm(directory, { recursive: true, force: true });
|
||||
}
|
||||
});
|
||||
|
||||
test("a replacing write that would stamp past year 9999 is refused and writes nothing", async () => {
|
||||
const directory = await scratch();
|
||||
try {
|
||||
const store = new SingleFileMessagePackBackingStore(join(directory, "store.cc"));
|
||||
const last = "9999-12-31T23:59:59.999Z";
|
||||
await store.push(entry("k", 1, last));
|
||||
await assert.rejects(() => store.push(entry("k", 2, last)), RangeError);
|
||||
assert.equal((await store.read("k"))!.envelope.storedAt, last);
|
||||
assert.deepEqual([...documentPayloadBytes((await store.read("k"))!.envelope.payload)], [0x91, 1]);
|
||||
} finally {
|
||||
await rm(directory, { recursive: true, force: true });
|
||||
}
|
||||
});
|
||||
|
||||
test("a write that restores an earlier value does not restore its version", async () => {
|
||||
const directory = await scratch();
|
||||
try {
|
||||
const store = new SingleFileMessagePackBackingStore(join(directory, "store.cc"));
|
||||
const at = "2026-03-04T05:06:07.100Z";
|
||||
await store.push(entry("k", 1, at));
|
||||
const first = (await store.read("k"))!.version;
|
||||
await store.push(entry("k", 5, at));
|
||||
await store.push(entry("k", 1, at));
|
||||
assert.notEqual((await store.read("k"))!.version, first, "an A-B-A within one millisecond keeps no version");
|
||||
assert.deepEqual(await store.commitIf(entry("k", 9, at), first), { outcome: "mismatch" });
|
||||
} finally {
|
||||
await rm(directory, { recursive: true, force: true });
|
||||
}
|
||||
});
|
||||
|
||||
test("two stores on one file never both commit against the same version", async () => {
|
||||
const directory = await scratch();
|
||||
try {
|
||||
const path = join(directory, "store.cc");
|
||||
const left = new SingleFileMessagePackBackingStore(path);
|
||||
const right = new SingleFileMessagePackBackingStore(path);
|
||||
await left.push(entry("k", 0));
|
||||
const version = (await left.read("k"))!.version;
|
||||
const outcomes = await Promise.all(Array.from({ length: 6 }, (_, i) => (i % 2 === 0 ? left : right).commitIf(entry("k", i + 1), version)));
|
||||
assert.equal(outcomes.filter((o) => o.outcome === "committed").length, 1);
|
||||
} finally {
|
||||
await rm(directory, { recursive: true, force: true });
|
||||
}
|
||||
});
|
||||
|
|
@ -424,7 +424,7 @@ test("a link put while a writer waits on the store lock is refused once it holds
|
|||
await new SingleFileMessagePackBackingStore(target).push(noteEnvelope("a"));
|
||||
const before = await readFile(target);
|
||||
const writes: Array<[string, (store: SingleFileMessagePackBackingStore) => Promise<void>]> = [
|
||||
["push", (store) => store.push(noteEnvelope("b"))],
|
||||
["push", async (store) => void (await store.push(noteEnvelope("b")))],
|
||||
["pushAll", (store) => store.pushAll([noteEnvelope("b")])],
|
||||
["delete", (store) => store.delete(noteEnvelope("a"))],
|
||||
];
|
||||
|
|
|
|||
|
|
@ -7,6 +7,7 @@ import { z } from "zod";
|
|||
import { codecs } from "../src";
|
||||
import type { Codec } from "../src/codec";
|
||||
import { defineDocumentType, documentPayloadFromBytes } from "../src/document";
|
||||
import { recordVersion } from "../src/single-file-messagepack-backing-store";
|
||||
import type { CacheBackingStore, CultCacheEnvelope } from "../src/types";
|
||||
|
||||
// The union-surface fixture: the codecs, schema and value tests/vectors/union-parity was written from, shared by the
|
||||
|
|
@ -235,6 +236,16 @@ export function memoryStore(entries: CultCacheEnvelope[]): CacheBackingStore & {
|
|||
async pullAll() { return entries.map((entry) => ({ ...entry })); },
|
||||
async push(entry) { entries.push(entry); },
|
||||
async delete() {},
|
||||
async read(key) {
|
||||
const envelope = [...entries].reverse().find((entry) => entry.key === key);
|
||||
return envelope === undefined ? undefined : { envelope, version: recordVersion(envelope) };
|
||||
},
|
||||
async commitIf(entry, expectedVersion) {
|
||||
const current = [...entries].reverse().find((candidate) => candidate.key === entry.key);
|
||||
if ((current === undefined ? null : recordVersion(current)) !== expectedVersion) return { outcome: "mismatch" };
|
||||
entries.push(entry);
|
||||
return { outcome: "committed", storedAt: entry.storedAt };
|
||||
},
|
||||
};
|
||||
}
|
||||
|
||||
|
|
|
|||
600
packages/cultcache-ts/test/update.test.ts
Normal file
600
packages/cultcache-ts/test/update.test.ts
Normal file
|
|
@ -0,0 +1,600 @@
|
|||
import assert from "node:assert/strict";
|
||||
import { mkdtemp, readFile, rm } from "node:fs/promises";
|
||||
import { tmpdir } from "node:os";
|
||||
import { join } from "node:path";
|
||||
import test from "node:test";
|
||||
import { setImmediate as tick } from "node:timers/promises";
|
||||
|
||||
import { decode, encode } from "@msgpack/msgpack";
|
||||
import { z } from "zod";
|
||||
|
||||
import { CultCache } from "../src/cult-cache";
|
||||
import { defineDocumentRegistry, defineDocumentType, documentPayloadBytes, documentPayloadFromBytes } from "../src/document";
|
||||
import { SingleFileMessagePackBackingStore, readSingleFileStore } from "../src/single-file-messagepack-backing-store";
|
||||
import { SchemaConflictError, schemaIdForEnvelope } from "../src/store-format";
|
||||
import type { CacheBackingStore } from "../src/types";
|
||||
import { envelopeOf, memoryStore, read, surfaceDocument } from "./union-surface-fixture";
|
||||
|
||||
// CultCache.update: read the record and its content version from the store, apply a function to a copy, commit only if the
|
||||
// version still holds, run again on a lost race (variants:cut_spec:cut-update-with-retry-ts.r1; the C# tests in
|
||||
// tests/GameCult.Caching.Tests/UpdateTests.cs are the parity).
|
||||
|
||||
const counterDocument = defineDocumentType({
|
||||
type: "update.counter",
|
||||
schemaId: "update.counter.v1",
|
||||
schemaName: "update.counter",
|
||||
schemaVersion: "update.counter.v1",
|
||||
schema: z.object({ value: z.number().int() }),
|
||||
});
|
||||
const tallyDocument = defineDocumentType({
|
||||
type: "update.tally",
|
||||
schemaId: "update.tally.v1",
|
||||
schemaName: "update.tally",
|
||||
schemaVersion: "update.tally.v1",
|
||||
schema: z.object({ count: z.number().int() }),
|
||||
});
|
||||
const registry = defineDocumentRegistry(counterDocument, tallyDocument);
|
||||
const settingsFields = { schemaName: "update.settings", schemaVersion: "update.settings.v1", schemaId: "update.settings.v1", schema: z.object({ theme: z.string() }) };
|
||||
const settingsDocument = defineDocumentType({ type: "update.settings", ...settingsFields, global: true });
|
||||
const plainSettingsDocument = defineDocumentType({ type: "update.settings", ...settingsFields });
|
||||
|
||||
const KEY = "counter";
|
||||
|
||||
async function withDirectory<T>(body: (path: string) => Promise<T>): Promise<T> {
|
||||
const directory = await mkdtemp(join(tmpdir(), "cultcache-update-"));
|
||||
try {
|
||||
return await body(join(directory, "store.cc"));
|
||||
} finally {
|
||||
await rm(directory, { recursive: true, force: true });
|
||||
}
|
||||
}
|
||||
|
||||
type Registry = Parameters<ReturnType<typeof CultCache.builder>["withRegistry"]>[0];
|
||||
|
||||
async function open(path: string, documents: Registry = registry): Promise<CultCache> {
|
||||
const cache = CultCache.builder().withRegistry(documents).withGenericStore(new SingleFileMessagePackBackingStore(path)).build();
|
||||
await cache.pullAllBackingStores();
|
||||
return cache;
|
||||
}
|
||||
|
||||
// What the file holds at the key, read without a cache.
|
||||
function stored(path: string, key = KEY): number | undefined {
|
||||
const envelope = readSingleFileStore(path).envelopes.find((candidate) => candidate.key === key);
|
||||
return envelope === undefined ? undefined : (decode(documentPayloadBytes(envelope.payload)) as { value: number }).value;
|
||||
}
|
||||
|
||||
const plus = (n: number) => (current: { value: number } | undefined) => ({ value: (current?.value ?? 0) + n });
|
||||
|
||||
test("two writers updating lose no increment", async () => {
|
||||
await withDirectory(async (path) => {
|
||||
const [left, right] = [await open(path), await open(path)];
|
||||
let calls = 0;
|
||||
const loop = async (cache: CultCache) => {
|
||||
const outcomes: string[] = [];
|
||||
for (let i = 0; i < 50; i++) {
|
||||
const result = await cache.update(counterDocument, KEY, async (current) => {
|
||||
calls++;
|
||||
await tick();
|
||||
return { value: (current?.value ?? 0) + 1 };
|
||||
}, { maxAttempts: 50 });
|
||||
outcomes.push(result.outcome);
|
||||
}
|
||||
return outcomes;
|
||||
};
|
||||
const outcomes = (await Promise.all([loop(left), loop(right)])).flat();
|
||||
assert.ok(outcomes.every((outcome) => outcome === "updated"), outcomes.join());
|
||||
assert.equal(stored(path), 100);
|
||||
assert.ok(calls > 100, `the caches raced: apply ran ${calls} times`);
|
||||
});
|
||||
});
|
||||
|
||||
test("a lost attempt reruns apply on the fresh value", async () => {
|
||||
await withDirectory(async (path) => {
|
||||
const [cache, rival] = [await open(path), await open(path)];
|
||||
await cache.put(counterDocument, KEY, { value: 1 });
|
||||
const seen: Array<number | undefined> = [];
|
||||
const result = await cache.update(counterDocument, KEY, async (current) => {
|
||||
seen.push(current?.value);
|
||||
if (seen.length === 1) await rival.put(counterDocument, KEY, { value: 100 });
|
||||
return { value: current!.value + 1 };
|
||||
});
|
||||
assert.deepEqual(seen, [1, 100]);
|
||||
assert.equal(result.outcome, "updated");
|
||||
assert.equal(result.document?.value, 101);
|
||||
assert.equal(stored(path), 101);
|
||||
});
|
||||
});
|
||||
|
||||
test("exhausted attempts return lostRace, and the cache then shows the store", async () => {
|
||||
await withDirectory(async (path) => {
|
||||
const [cache, rival] = [await open(path), await open(path)];
|
||||
await cache.put(counterDocument, KEY, { value: 1 });
|
||||
let calls = 0;
|
||||
const result = await cache.update(
|
||||
counterDocument,
|
||||
KEY,
|
||||
async (current) => {
|
||||
calls++;
|
||||
await rival.put(counterDocument, KEY, { value: 100 * calls });
|
||||
return { value: current!.value + 1 };
|
||||
},
|
||||
{ maxAttempts: 3 },
|
||||
);
|
||||
assert.equal(calls, 3);
|
||||
assert.equal(result.outcome, "lostRace");
|
||||
assert.equal(result.document?.value, 300);
|
||||
assert.equal(stored(path), 300);
|
||||
});
|
||||
});
|
||||
|
||||
test("five attempts by default", async () => {
|
||||
await withDirectory(async (path) => {
|
||||
const [cache, rival] = [await open(path), await open(path)];
|
||||
let calls = 0;
|
||||
const result = await cache.update(counterDocument, KEY, async () => {
|
||||
calls++;
|
||||
await rival.put(counterDocument, KEY, { value: calls });
|
||||
return { value: -1 };
|
||||
});
|
||||
assert.equal(result.outcome, "lostRace");
|
||||
assert.equal(calls, 5);
|
||||
});
|
||||
});
|
||||
|
||||
test("maxAttempts 1 makes one call", async () => {
|
||||
await withDirectory(async (path) => {
|
||||
const [cache, rival] = [await open(path), await open(path)];
|
||||
await cache.put(counterDocument, KEY, { value: 1 });
|
||||
let calls = 0;
|
||||
const apply = async (current: { value: number } | undefined) => {
|
||||
calls++;
|
||||
await rival.put(counterDocument, KEY, { value: 50 + calls });
|
||||
return { value: current!.value + 1 };
|
||||
};
|
||||
assert.equal((await cache.update(counterDocument, KEY, apply, { maxAttempts: 1 })).outcome, "lostRace");
|
||||
assert.equal(calls, 1);
|
||||
});
|
||||
});
|
||||
|
||||
// An apply that declines resolves at once whatever the bound, so a validation that lets a bad value through fails the assertion
|
||||
// below instead of looping: no apply here ever loses a race.
|
||||
test("invalid maxAttempts (undefined is the default, null is refused) and a non-function reject before any call", async () => {
|
||||
await withDirectory(async (path) => {
|
||||
const cache = await open(path);
|
||||
let calls = 0;
|
||||
const apply = () => (calls++, undefined);
|
||||
for (const maxAttempts of [0, -1, 1.5, Number.NaN, Number.POSITIVE_INFINITY, null as never]) {
|
||||
await assert.rejects(() => cache.update(counterDocument, KEY, apply, { maxAttempts }), RangeError, `maxAttempts ${String(maxAttempts)}`);
|
||||
}
|
||||
await assert.rejects(() => cache.update(counterDocument, KEY, "no" as never), TypeError);
|
||||
assert.equal(calls, 0);
|
||||
assert.equal((await cache.update(counterDocument, KEY, apply, { maxAttempts: undefined })).outcome, "declined");
|
||||
assert.equal(calls, 1);
|
||||
});
|
||||
});
|
||||
|
||||
test("null from apply declines as undefined does: nothing is written", async () => {
|
||||
await withDirectory(async (path) => {
|
||||
const cache = await open(path);
|
||||
await cache.put(counterDocument, KEY, { value: 1 });
|
||||
const before = await readFile(path);
|
||||
const result = await cache.update(counterDocument, KEY, () => null as never);
|
||||
assert.deepEqual(result, { outcome: "declined", document: cache.get(counterDocument, KEY) });
|
||||
assert.deepEqual([...(await readFile(path))], [...before]);
|
||||
});
|
||||
});
|
||||
|
||||
test("a cache with no store refuses a key held as another type before apply runs", async () => {
|
||||
const cache = CultCache.builder().withRegistry(registry).build();
|
||||
await cache.put(tallyDocument, KEY, { count: 1 });
|
||||
let calls = 0;
|
||||
await assert.rejects(() => cache.update(counterDocument, KEY, () => (calls++, { value: 1 })), /is not a "update.counter"/);
|
||||
assert.equal(calls, 0);
|
||||
});
|
||||
|
||||
test("a store attached while apply awaits turns the commit into a lost race, never a blind push", async () => {
|
||||
await withDirectory(async (path) => {
|
||||
const writer = await open(path);
|
||||
await writer.put(counterDocument, KEY, { value: 50 });
|
||||
const cache = CultCache.builder().withRegistry(registry).build();
|
||||
const seen: Array<number | undefined> = [];
|
||||
const result = await cache.update(counterDocument, KEY, async (current) => {
|
||||
seen.push(current?.value);
|
||||
if (seen.length === 1) cache.addBackingStore(new SingleFileMessagePackBackingStore(path));
|
||||
return { value: (current?.value ?? 0) + 1 };
|
||||
});
|
||||
assert.deepEqual(seen, [undefined, 50], "the second attempt reads the store's record");
|
||||
assert.equal(result.outcome, "updated");
|
||||
assert.equal(stored(path), 51);
|
||||
});
|
||||
});
|
||||
|
||||
test("put installs the storedAt the store stamped, so the cache and the file agree", async () => {
|
||||
await withDirectory(async (path) => {
|
||||
const cache = await open(path);
|
||||
await cache.put(counterDocument, KEY, { value: 1 });
|
||||
const store = new SingleFileMessagePackBackingStore(path);
|
||||
const [envelope] = readSingleFileStore(path).envelopes;
|
||||
await store.push({ ...envelope!, payload: documentPayloadFromBytes(encode({ value: 99 })), storedAt: "2099-01-01T00:00:00.000Z" });
|
||||
await cache.put(counterDocument, KEY, { value: 2 });
|
||||
const onDisk = readSingleFileStore(path).envelopes[0]!.storedAt;
|
||||
assert.equal(onDisk, "2099-01-01T00:00:00.001Z");
|
||||
assert.equal(cache.getEnvelope(counterDocument, KEY)!.storedAt, onDisk);
|
||||
});
|
||||
});
|
||||
|
||||
test("apply gets a fresh copy: changing it changes nothing held, and a lost attempt's change is not seen by the next", async () => {
|
||||
await withDirectory(async (path) => {
|
||||
const [cache, rival] = [await open(path), await open(path)];
|
||||
await cache.put(counterDocument, KEY, { value: 1 });
|
||||
const argument: Array<{ value: number } | undefined> = [];
|
||||
const seen: number[] = [];
|
||||
await cache.update(counterDocument, KEY, async (current) => {
|
||||
argument.push(current);
|
||||
assert.notEqual(current, cache.get(counterDocument, KEY));
|
||||
seen.push(current!.value);
|
||||
current!.value += 1000;
|
||||
assert.equal(cache.get(counterDocument, KEY)!.value, 1);
|
||||
if (seen.length === 1) await rival.put(counterDocument, KEY, { value: 7 });
|
||||
return { value: current!.value + 1 };
|
||||
});
|
||||
assert.deepEqual(seen, [1, 7], "the second attempt does not see the first one's change");
|
||||
assert.notEqual(argument[0], argument[1]);
|
||||
});
|
||||
});
|
||||
|
||||
test("an in-place change to the held value is neither applied nor persisted", async () => {
|
||||
await withDirectory(async (path) => {
|
||||
const [cache, fresh] = [await open(path), await open(path)];
|
||||
await cache.put(counterDocument, KEY, { value: 1 });
|
||||
cache.get(counterDocument, KEY)!.value = 999;
|
||||
const seen: number[] = [];
|
||||
const result = await cache.update(counterDocument, KEY, (current) => {
|
||||
seen.push(current!.value);
|
||||
return { value: current!.value + 1 };
|
||||
});
|
||||
assert.deepEqual(seen, [1]);
|
||||
assert.equal(stored(path), 2);
|
||||
await fresh.pullAllBackingStores();
|
||||
assert.equal(fresh.get(counterDocument, KEY)!.value, 2);
|
||||
result.document!.value = 999;
|
||||
await cache.update(counterDocument, KEY, plus(1));
|
||||
assert.equal(stored(path), 3);
|
||||
});
|
||||
});
|
||||
|
||||
test("undefined from apply declines: nothing is written", async () => {
|
||||
await withDirectory(async (path) => {
|
||||
const cache = await open(path);
|
||||
await cache.put(counterDocument, KEY, { value: 1 });
|
||||
const before = await readFile(path);
|
||||
const held = await cache.update(counterDocument, KEY, () => undefined);
|
||||
assert.equal(held.outcome, "declined");
|
||||
assert.equal(held.document, cache.get(counterDocument, KEY));
|
||||
const absent = await cache.update(counterDocument, "absent", () => undefined);
|
||||
assert.deepEqual(absent, { outcome: "declined", document: undefined });
|
||||
assert.deepEqual([...(await readFile(path))], [...before]);
|
||||
});
|
||||
});
|
||||
|
||||
test("an absent key is created once under a race", async () => {
|
||||
await withDirectory(async (path) => {
|
||||
const [left, right] = [await open(path), await open(path)];
|
||||
const outcomes = await Promise.all(
|
||||
[left, right].map((cache) =>
|
||||
cache.update(counterDocument, KEY, async (current) => {
|
||||
await tick();
|
||||
return { value: (current?.value ?? 0) + 1 };
|
||||
}),
|
||||
),
|
||||
);
|
||||
assert.deepEqual(outcomes.map((result) => result.outcome), ["updated", "updated"]);
|
||||
assert.equal(stored(path), 2);
|
||||
});
|
||||
});
|
||||
|
||||
test("a rival writing and restoring the value still reruns apply", async () => {
|
||||
await withDirectory(async (path) => {
|
||||
const [cache, rival] = [await open(path), await open(path)];
|
||||
await cache.put(counterDocument, KEY, { value: 1 });
|
||||
const seen: number[] = [];
|
||||
await cache.update(counterDocument, KEY, async (current) => {
|
||||
seen.push(current!.value);
|
||||
if (seen.length === 1) {
|
||||
await rival.put(counterDocument, KEY, { value: 5 });
|
||||
await rival.put(counterDocument, KEY, { value: 1 });
|
||||
}
|
||||
return { value: current!.value + 1 };
|
||||
});
|
||||
assert.deepEqual(seen, [1, 1]);
|
||||
assert.equal(stored(path), 2);
|
||||
});
|
||||
});
|
||||
|
||||
test("a rival removal is seen as undefined", async () => {
|
||||
await withDirectory(async (path) => {
|
||||
const [cache, rival] = [await open(path), await open(path)];
|
||||
await cache.put(counterDocument, KEY, { value: 1 });
|
||||
const seen: Array<number | undefined> = [];
|
||||
await cache.update(counterDocument, KEY, async (current) => {
|
||||
seen.push(current?.value);
|
||||
if (seen.length === 1) {
|
||||
await rival.pullAllBackingStores();
|
||||
await rival.delete(counterDocument, KEY);
|
||||
}
|
||||
return { value: (current?.value ?? 0) + 1 };
|
||||
});
|
||||
assert.deepEqual(seen, [1, undefined]);
|
||||
assert.equal(stored(path), 1);
|
||||
});
|
||||
});
|
||||
|
||||
test("a stale cache's first attempt reads the store", async () => {
|
||||
await withDirectory(async (path) => {
|
||||
const [stale, rival] = [await open(path), await open(path)];
|
||||
await rival.put(counterDocument, KEY, { value: 1 });
|
||||
await stale.pullAllBackingStores();
|
||||
await rival.put(counterDocument, KEY, { value: 7 });
|
||||
const seen: number[] = [];
|
||||
await stale.update(counterDocument, KEY, (current) => {
|
||||
seen.push(current!.value);
|
||||
return { value: current!.value + 1 };
|
||||
});
|
||||
assert.deepEqual(seen, [7]);
|
||||
assert.equal(stored(path), 8);
|
||||
});
|
||||
});
|
||||
|
||||
test("refusals are not retried and apply is not called for a record it cannot copy", async () => {
|
||||
await withDirectory(async (path) => {
|
||||
const cache = await open(path);
|
||||
await cache.put(tallyDocument, "tally", { count: 1 });
|
||||
const before = await readFile(path);
|
||||
let calls = 0;
|
||||
const apply = (current: { value: number } | undefined) => {
|
||||
calls++;
|
||||
return { value: (current?.value ?? 0) + 1 };
|
||||
};
|
||||
// Another type at the key.
|
||||
await assert.rejects(() => cache.update(counterDocument, "tally", apply), /not a "update.counter"/);
|
||||
// A foreign record: a type this cache has no definition for.
|
||||
const narrow = await open(path, defineDocumentRegistry(counterDocument));
|
||||
await assert.rejects(() => narrow.update(counterDocument, "tally", apply), /not a "update.counter"/);
|
||||
assert.equal(calls, 0);
|
||||
assert.deepEqual([...(await readFile(path))], [...before]);
|
||||
|
||||
// A store without read and commitIf cannot take a conditional write.
|
||||
const bare: CacheBackingStore = { pullAll: async () => [], push: async () => {}, delete: async () => {} };
|
||||
const plain = CultCache.builder().withRegistry(registry).withGenericStore(bare).build();
|
||||
await plain.pullAllBackingStores();
|
||||
await assert.rejects(() => plain.update(counterDocument, KEY, apply), /read and commitIf/);
|
||||
assert.equal(calls, 0);
|
||||
|
||||
// A record held read-only is refused by the judge after apply ran once; apply that throws is not retried either.
|
||||
const surface = surfaceDocument();
|
||||
const entries = [envelopeOf(await read("extra-slot-record.msgpack"))];
|
||||
const readOnly = CultCache.builder().withDocumentType(surface).withGenericStore(memoryStore(entries)).build();
|
||||
await readOnly.pullAllBackingStores();
|
||||
let surfaceCalls = 0;
|
||||
await assert.rejects(() => readOnly.update(surface, "surface", (value) => (surfaceCalls++, value)), SchemaConflictError);
|
||||
assert.equal(surfaceCalls, 1);
|
||||
assert.equal(entries.length, 1, "nothing reached the store");
|
||||
await assert.rejects(
|
||||
() => cache.update(counterDocument, KEY, () => { calls++; throw new Error("boom"); }),
|
||||
/boom/,
|
||||
);
|
||||
assert.equal(calls, 1);
|
||||
});
|
||||
});
|
||||
|
||||
test("own writes cause no retry, and updated returns the held instance", async () => {
|
||||
await withDirectory(async (path) => {
|
||||
const cache = await open(path);
|
||||
await cache.put(counterDocument, KEY, { value: 1 });
|
||||
let calls = 0;
|
||||
const apply = (current: { value: number } | undefined) => {
|
||||
calls++;
|
||||
return { value: current!.value + 1 };
|
||||
};
|
||||
await cache.update(counterDocument, KEY, apply);
|
||||
const second = await cache.update(counterDocument, KEY, apply);
|
||||
assert.equal(calls, 2);
|
||||
assert.equal(second.outcome, "updated");
|
||||
assert.equal(second.document, cache.get(counterDocument, KEY));
|
||||
assert.equal(stored(path), 3);
|
||||
// The cache holds the record as the store wrote it, the stamp the store chose included: over a record stamped in the future
|
||||
// the store writes one millisecond after it, whatever the clock says.
|
||||
const future = readSingleFileStore(path).envelopes[0]!;
|
||||
await new SingleFileMessagePackBackingStore(path).push({ ...future, payload: documentPayloadFromBytes(encode({ value: 99 })), storedAt: "2099-01-01T00:00:00.000Z" });
|
||||
await cache.update(counterDocument, KEY, apply);
|
||||
assert.equal(readSingleFileStore(path).envelopes[0]!.storedAt, "2099-01-01T00:00:00.001Z");
|
||||
assert.equal(cache.getEnvelope(counterDocument, KEY)!.storedAt, "2099-01-01T00:00:00.001Z");
|
||||
});
|
||||
});
|
||||
|
||||
test("apply runs outside the cache's serial queue", async () => {
|
||||
await withDirectory(async (path) => {
|
||||
const cache = await open(path);
|
||||
let calls = 0;
|
||||
const result = await cache.update(counterDocument, KEY, async (current) => {
|
||||
calls++;
|
||||
await Promise.race([
|
||||
cache.put(tallyDocument, "other", { count: 1 }),
|
||||
new Promise((_, reject) => setTimeout(() => reject(new Error("a cache call from apply did not finish")), 5000).unref()),
|
||||
]);
|
||||
return { value: (current?.value ?? 0) + 1 };
|
||||
});
|
||||
assert.equal(result.outcome, "updated");
|
||||
assert.equal(calls, 1);
|
||||
});
|
||||
});
|
||||
|
||||
test("a cache with no store loses no increment under interleaved updates", async () => {
|
||||
const cache = CultCache.builder().withRegistry(registry).build();
|
||||
await cache.pullAllBackingStores();
|
||||
let calls = 0;
|
||||
const loop = async () => {
|
||||
let updated = 0;
|
||||
for (let i = 0; i < 50; i++) {
|
||||
const result = await cache.update(counterDocument, KEY, async (current) => {
|
||||
calls++;
|
||||
await tick();
|
||||
return { value: (current?.value ?? 0) + 1 };
|
||||
});
|
||||
if (result.outcome === "updated") updated++;
|
||||
}
|
||||
return updated;
|
||||
};
|
||||
// The two loops run in lockstep, so one may lose every round (a lostRace is the answer, not a lost write): what is held is
|
||||
// exactly what the updated outcomes wrote.
|
||||
const updated = (await Promise.all([loop(), loop()])).reduce((sum, n) => sum + n, 0);
|
||||
assert.equal(cache.get(counterDocument, KEY)!.value, updated);
|
||||
assert.ok(calls > updated, `the loops raced: apply ran ${calls} times for ${updated} writes`);
|
||||
});
|
||||
|
||||
test("a store-less cache copies the bytes it admitted, not the held value", async () => {
|
||||
const cache = CultCache.builder().withRegistry(registry).build();
|
||||
await cache.put(counterDocument, KEY, { value: 1 });
|
||||
cache.get(counterDocument, KEY)!.value = 999;
|
||||
const seen: number[] = [];
|
||||
await cache.update(counterDocument, KEY, (current) => (seen.push(current!.value), { value: current!.value + 1 }));
|
||||
assert.deepEqual(seen, [1]);
|
||||
});
|
||||
|
||||
test("updateGlobal updates the global record and refuses one loaded from a legacy key until putGlobal retires it", async () => {
|
||||
await withDirectory(async (path) => {
|
||||
const settings = defineDocumentRegistry(settingsDocument);
|
||||
const cache = await open(path, settings);
|
||||
const first = await cache.updateGlobal(settingsDocument, (current) => ({ theme: `${current?.theme ?? ""}a` }));
|
||||
assert.equal(first.outcome, "updated");
|
||||
assert.deepEqual(cache.getGlobal(settingsDocument), { theme: "a" });
|
||||
assert.equal(readSingleFileStore(path).envelopes[0]!.key, CultCache.GLOBAL_KEY);
|
||||
await rm(path);
|
||||
|
||||
const legacy = await open(path, defineDocumentRegistry(plainSettingsDocument));
|
||||
await legacy.put(plainSettingsDocument, "old-settings", { theme: "x" });
|
||||
const loaded = await open(path, settings);
|
||||
let calls = 0;
|
||||
await assert.rejects(() => loaded.updateGlobal(settingsDocument, (current) => (calls++, current)), /putGlobal to retire its legacy key/);
|
||||
assert.equal(calls, 0);
|
||||
// Every path to the write is refused, the global's own key included, and the legacy record survives.
|
||||
const before = await readFile(path);
|
||||
await assert.rejects(() => loaded.update(settingsDocument, CultCache.GLOBAL_KEY, () => ({ theme: "lost" })), /putGlobal to retire its legacy key/);
|
||||
assert.deepEqual([...(await readFile(path))], [...before]);
|
||||
assert.deepEqual(readSingleFileStore(path).envelopes.map((e) => e.key), ["old-settings"]);
|
||||
await loaded.putGlobal(settingsDocument, { theme: "y" });
|
||||
assert.equal((await loaded.updateGlobal(settingsDocument, (current) => ({ theme: `${current!.theme}z` }))).outcome, "updated");
|
||||
assert.deepEqual(loaded.getGlobal(settingsDocument), { theme: "yz" });
|
||||
await assert.rejects(() => cache.update(settingsDocument, "not-global", () => ({ theme: "q" })), /must use key/);
|
||||
});
|
||||
});
|
||||
|
||||
test("a legacy global that a pull loads while apply awaits is refused at the write, and its legacy record survives", async () => {
|
||||
await withDirectory(async (path) => {
|
||||
const settings = defineDocumentRegistry(settingsDocument);
|
||||
const cache = await open(path, settings);
|
||||
const legacy = await open(path, defineDocumentRegistry(plainSettingsDocument));
|
||||
await assert.rejects(
|
||||
() =>
|
||||
cache.update(settingsDocument, CultCache.GLOBAL_KEY, async () => {
|
||||
await legacy.put(plainSettingsDocument, "old-settings", { theme: "x" });
|
||||
await cache.pullAllBackingStores();
|
||||
return { theme: "lost" };
|
||||
}),
|
||||
/putGlobal to retire its legacy key/,
|
||||
);
|
||||
assert.deepEqual(readSingleFileStore(path).envelopes.map((e) => e.key), ["old-settings"]);
|
||||
});
|
||||
});
|
||||
|
||||
test("the ledger holds under contention: the stored value is the number of updated outcomes", async () => {
|
||||
await withDirectory(async (path) => {
|
||||
const caches = await Promise.all([open(path), open(path), open(path)]);
|
||||
const results = await Promise.all(
|
||||
caches.map(async (cache) => {
|
||||
let updated = 0;
|
||||
for (let i = 0; i < 25; i++) {
|
||||
const result = await cache.update(counterDocument, KEY, async (current) => {
|
||||
await tick();
|
||||
return { value: (current?.value ?? 0) + 1 };
|
||||
});
|
||||
if (result.outcome === "updated") updated++;
|
||||
}
|
||||
return updated;
|
||||
}),
|
||||
);
|
||||
assert.equal(stored(path), results.reduce((sum, n) => sum + n, 0));
|
||||
});
|
||||
});
|
||||
|
||||
test("a flush of an unchanged cache rewrites the file byte for byte, and an update does not lose a race to it", async () => {
|
||||
await withDirectory(async (path) => {
|
||||
const cache = await open(path);
|
||||
await cache.put(counterDocument, KEY, { value: 1 });
|
||||
await cache.put(counterDocument, "other", { value: 7 });
|
||||
const store = new SingleFileMessagePackBackingStore(path);
|
||||
const before = await readFile(path);
|
||||
// What CultMesh.flush does: push the whole stored image, twice.
|
||||
await store.pushAll(await store.pullAll());
|
||||
await store.pushAll(await store.pullAll());
|
||||
assert.deepEqual(await readFile(path), before);
|
||||
const result = await cache.update(counterDocument, KEY, async (current) => {
|
||||
await store.pushAll(await store.pullAll());
|
||||
return { value: current!.value + 1 };
|
||||
}, { maxAttempts: 1 });
|
||||
assert.equal(result.outcome, "updated");
|
||||
assert.equal(stored(path), 2);
|
||||
});
|
||||
});
|
||||
|
||||
test("a changed record is re-stamped later than the one it replaces, and a revert to earlier bytes too", async () => {
|
||||
await withDirectory(async (path) => {
|
||||
const store = new SingleFileMessagePackBackingStore(path);
|
||||
const cache = await open(path);
|
||||
await cache.put(counterDocument, KEY, { value: 1 });
|
||||
const first = readSingleFileStore(path).envelopes[0]!;
|
||||
const same = await store.push({ ...first, storedAt: "2099-01-01T00:00:00.000Z" });
|
||||
assert.equal(same.storedAt, first.storedAt, "identical content keeps its stamp");
|
||||
await cache.put(counterDocument, KEY, { value: 2 });
|
||||
const second = readSingleFileStore(path).envelopes[0]!;
|
||||
const reverted = await store.push({ ...first, storedAt: second.storedAt });
|
||||
assert.ok(Date.parse(reverted.storedAt) > Date.parse(second.storedAt));
|
||||
});
|
||||
});
|
||||
|
||||
test("a push result that is not a parseable stamp is ignored", async () => {
|
||||
for (const result of [true, {}, "", "garbage", { storedAt: "" }, { storedAt: "garbage" }]) {
|
||||
const entries: Parameters<typeof memoryStore>[0] = [];
|
||||
const store = { ...memoryStore(entries), push: async (entry: (typeof entries)[number]) => { entries.push(entry); return result as never; } };
|
||||
const cache = CultCache.builder().withRegistry(registry).withGenericStore(store).build();
|
||||
await cache.put(counterDocument, KEY, { value: 1 });
|
||||
const held = cache.getEnvelope(counterDocument, KEY)!.storedAt;
|
||||
assert.equal(Number.isNaN(Date.parse(held)), false, `held stamp ${held} after a push result of ${JSON.stringify(result)}`);
|
||||
}
|
||||
});
|
||||
|
||||
test("a rewrite of identical content keeps the stamp but writes the writer's type and catalog entry", async () => {
|
||||
await withDirectory(async (path) => {
|
||||
const old = defineDocumentType({ type: "update.rn.old", schemaId: "update.rn.v1", schemaName: "update.rn.old", schemaVersion: "update.rn.v1", schema: z.object({ value: z.number().int() }) });
|
||||
const renamed = defineDocumentType({ type: "update.rn.new", schemaId: "update.rn.v1", schemaName: "update.rn.new", schemaVersion: "update.rn.v1", schema: z.object({ value: z.number().int() }) });
|
||||
await (await open(path, defineDocumentRegistry(old))).put(old, KEY, { value: 1 });
|
||||
const before = readSingleFileStore(path).envelopes[0]!;
|
||||
await (await open(path, defineDocumentRegistry(renamed))).put(renamed, KEY, { value: 1 });
|
||||
const after = readSingleFileStore(path).envelopes[0]!;
|
||||
assert.equal(after.storedAt, before.storedAt, "identical content keeps its stamp");
|
||||
assert.equal(after.type, "update.rn.new", "the writer's type is what the file says");
|
||||
});
|
||||
});
|
||||
|
||||
test("a rewrite of the same bytes under another schema id is a changed record: stamped later, written under the new id", async () => {
|
||||
await withDirectory(async (path) => {
|
||||
const v1 = defineDocumentType({ type: "update.sid", schemaId: "update.sid.v1", schemaName: "update.sid", schemaVersion: "update.sid.v1", schema: z.object({ value: z.number().int() }) });
|
||||
const v2 = defineDocumentType({ type: "update.sid", schemaId: "update.sid.v2", schemaName: "update.sid", schemaVersion: "update.sid.v2", compatibleSchemaIds: ["update.sid.v1"], schema: z.object({ value: z.number().int() }) });
|
||||
await (await open(path, defineDocumentRegistry(v1))).put(v1, KEY, { value: 1 });
|
||||
const before = readSingleFileStore(path).envelopes[0]!;
|
||||
await (await open(path, defineDocumentRegistry(v2))).put(v2, KEY, { value: 1 });
|
||||
const after = readSingleFileStore(path).envelopes[0]!;
|
||||
assert.ok(Date.parse(after.storedAt) > Date.parse(before.storedAt));
|
||||
assert.equal(schemaIdForEnvelope(after), "update.sid.v2");
|
||||
});
|
||||
});
|
||||
115
tests/GameCult.Caching.Tests/ContentVersionVectorTests.cs
Normal file
115
tests/GameCult.Caching.Tests/ContentVersionVectorTests.cs
Normal file
|
|
@ -0,0 +1,115 @@
|
|||
#nullable enable
|
||||
using System;
|
||||
using System.Buffers;
|
||||
using System.IO;
|
||||
using System.Linq;
|
||||
using GameCult.Caching;
|
||||
using GameCult.Caching.MessagePack;
|
||||
using MessagePack;
|
||||
using NUnit.Framework;
|
||||
|
||||
namespace GameCult.Caching.Tests
|
||||
{
|
||||
// tests/vectors/content-version: the content version of a stored record, the lowercase hex SHA-256 of its persisted encoding
|
||||
// (variants:ruling:write-set-content-version-declared-ownership), as the reference reads it from a single-file store. The C#
|
||||
// reference is the only writer: under CULTCACHE_WRITE_VECTORS=1 the store is written and the versions read from it are written
|
||||
// beside it; every other run opens the committed store.cc and compares what the reference reads to versions.msgpack. Another
|
||||
// runtime computes the version from the same file and must equal these rows.
|
||||
public class ContentVersionVectorTests
|
||||
{
|
||||
private const string Vector = "content-version";
|
||||
|
||||
private static readonly Type Item =
|
||||
EmittedDocumentTypes.Emit("CvItem", "tests.cv_item", "tests.cv_item.v1", new[] { new EmittedDocumentTypes.Field("Text", typeof(string), 0) });
|
||||
|
||||
private static readonly CultDocumentRegistry Registry = CultDocumentRegistry.ForTypes(new[] { Item });
|
||||
|
||||
private static readonly (string Key, string Text)[] Records = { ("alpha", "first"), ("beta", "second") };
|
||||
|
||||
private static byte[] StoreBytes()
|
||||
{
|
||||
var directory = Path.Combine(Path.GetTempPath(), $"cultlib-cvv-{Guid.NewGuid():N}");
|
||||
Directory.CreateDirectory(directory);
|
||||
try
|
||||
{
|
||||
var path = Path.Combine(directory, "store.cc");
|
||||
using (var cache = CultCacheMessagePack.Create(path, new CultCacheOpenOptions { Registry = Registry, StoreFlushOnDispose = true }))
|
||||
{
|
||||
cache.Clock = () => VectorFiles.StoredAt;
|
||||
foreach (var (key, text) in Records)
|
||||
cache.UpsertAsync(Item, EmittedDocumentTypes.New(Item, ("Text", text)), new CultRecordKey(key)).GetAwaiter().GetResult();
|
||||
cache.FlushAsync().GetAwaiter().GetResult();
|
||||
}
|
||||
|
||||
return File.ReadAllBytes(path);
|
||||
}
|
||||
finally
|
||||
{
|
||||
Directory.Delete(directory, recursive: true);
|
||||
}
|
||||
}
|
||||
|
||||
// What the reference reads from a store file: [key, version] for each record, sorted by key.
|
||||
private static (string Key, string Version)[] ReadVersions(byte[] store)
|
||||
{
|
||||
var directory = Path.Combine(Path.GetTempPath(), $"cultlib-cvr-{Guid.NewGuid():N}");
|
||||
Directory.CreateDirectory(directory);
|
||||
try
|
||||
{
|
||||
var path = Path.Combine(directory, "store.cc");
|
||||
File.WriteAllBytes(path, store);
|
||||
using var cache = CultCacheMessagePack.Create(path, new CultCacheOpenOptions { Registry = Registry });
|
||||
return cache.AllStoredDocuments
|
||||
.Select(stored => (Key: stored.Key.Value, Version: stored.StoredVersion))
|
||||
.OrderBy(row => row.Key, StringComparer.Ordinal)
|
||||
.ToArray();
|
||||
}
|
||||
finally
|
||||
{
|
||||
Directory.Delete(directory, recursive: true);
|
||||
}
|
||||
}
|
||||
|
||||
private static byte[] RowsFile((string Key, string Version)[] rows)
|
||||
{
|
||||
var buffer = new ArrayBufferWriter<byte>();
|
||||
var writer = new MessagePackWriter(buffer);
|
||||
writer.WriteArrayHeader(rows.Length);
|
||||
foreach (var (key, version) in rows)
|
||||
{
|
||||
writer.WriteArrayHeader(2);
|
||||
writer.Write(key);
|
||||
writer.Write(version);
|
||||
}
|
||||
|
||||
writer.Flush();
|
||||
return buffer.WrittenSpan.ToArray();
|
||||
}
|
||||
|
||||
[Test]
|
||||
public void TheReferenceReadsTheCommittedVersions()
|
||||
{
|
||||
if (VectorFiles.Writing)
|
||||
{
|
||||
var store = StoreBytes();
|
||||
VectorFiles.Check(Vector, "store.cc", store);
|
||||
VectorFiles.Check(Vector, "versions.msgpack", RowsFile(ReadVersions(store)));
|
||||
return;
|
||||
}
|
||||
|
||||
var committed = VectorFiles.Committed(Vector, "store.cc");
|
||||
var rows = ReadVersions(committed);
|
||||
Assert.That(rows.Select(row => row.Key), Is.EqualTo(Records.Select(record => record.Key)), "the store holds the two records");
|
||||
Assert.That(rows.Select(row => row.Version).Distinct().Count(), Is.EqualTo(rows.Length), "two different records, two versions");
|
||||
Assert.That(rows.All(row => row.Version.Length == 64 && row.Version == row.Version.ToLowerInvariant()), Is.True, "lowercase hex SHA-256");
|
||||
Assert.That(Convert.ToHexString(RowsFile(rows)), Is.EqualTo(Convert.ToHexString(VectorFiles.Committed(Vector, "versions.msgpack"))), "versions.msgpack");
|
||||
}
|
||||
|
||||
[Test]
|
||||
public void TheCommittedStoreIsWhatTheReferenceWrites()
|
||||
{
|
||||
if (VectorFiles.Writing) return;
|
||||
Assert.That(Convert.ToHexString(StoreBytes()), Is.EqualTo(Convert.ToHexString(VectorFiles.Committed(Vector, "store.cc"))), "store.cc");
|
||||
}
|
||||
}
|
||||
}
|
||||
17
tests/vectors/content-version/README.md
Normal file
17
tests/vectors/content-version/README.md
Normal file
|
|
@ -0,0 +1,17 @@
|
|||
# content-version vectors
|
||||
|
||||
Written only by the C# reference, `tests/GameCult.Caching.Tests/ContentVersionVectorTests.cs`, with
|
||||
`CULTCACHE_WRITE_VECTORS=1` (never set in CI). Every other run opens the committed `store.cc`, compares what the reference reads
|
||||
to `versions.msgpack` and compares the store bytes the reference writes to the file; a changed vector is a reviewed commit.
|
||||
Other runtimes compute the version of each record of `store.cc` and must equal these rows
|
||||
(`variants:ruling:write-set-content-version-declared-ownership`).
|
||||
|
||||
RULE: the content version of a stored record is the lowercase hex SHA-256 of its persisted encoding, the MessagePack array
|
||||
`[key, schemaId, storedAt, payload bytes]` exactly as the store writes the record.
|
||||
|
||||
## Files
|
||||
|
||||
- `store.cc`: a single-file store of one type, `tests.cv_item` (one string slot, `[0: Text]`), holding `alpha` = "first" and
|
||||
`beta` = "second", stored-at pinned to `2026-01-01T00:00:00Z`.
|
||||
- `versions.msgpack`: an array of `[key, version]`, sorted by key, the version the reference reads for the record
|
||||
(`GetStored(key).StoredVersion`).
|
||||
BIN
tests/vectors/content-version/store.cc
Normal file
BIN
tests/vectors/content-version/store.cc
Normal file
Binary file not shown.
1
tests/vectors/content-version/versions.msgpack
Normal file
1
tests/vectors/content-version/versions.msgpack
Normal file
|
|
@ -0,0 +1 @@
|
|||
’’¥alphaÙ@d330976df4f3946d0ad590d2091666aed4d5ebeac5d473792b4ef401b25c92fd’¤betaÙ@878026249c0df9a969ac1cac67bcdaab61f1b085ce4916c3c63f0fbef8f6da9f
|
||||
Loading…
Reference in a new issue