Carry portable schema identity on raw records
This commit is contained in:
parent
a2301b6b09
commit
c5d9435deb
6 changed files with 40 additions and 4 deletions
|
|
@ -571,7 +571,7 @@ export class CultMeshBrowserClient implements AsyncDisposable {
|
|||
this.#pendingSnapshots.delete(message.messageId);
|
||||
clearTimeout(pending.timer);
|
||||
const matches = message.documents.filter(record =>
|
||||
record.schemaId === pending.lease.schemaId && record.recordKey === pending.lease.recordKey);
|
||||
record.recordKey === pending.lease.recordKey && recordMatchesSchema(record, pending.lease.schemaId));
|
||||
if (matches.length > 1) {
|
||||
pending.reject(new Error(`CultMesh snapshot returned duplicate record '${pending.lease.recordKey}'.`));
|
||||
return;
|
||||
|
|
@ -584,11 +584,11 @@ export class CultMeshBrowserClient implements AsyncDisposable {
|
|||
const lease = this.#leases.get(message.subscriptionId) ?? this.#openingLeases.get(message.subscriptionId);
|
||||
if (!lease) return;
|
||||
if (message.changeKind === "removed") {
|
||||
if (message.recordKey === lease.recordKey && message.schemaId === lease.schemaId) lease.apply(undefined);
|
||||
if (message.recordKey === lease.recordKey) lease.apply(undefined);
|
||||
return;
|
||||
}
|
||||
const record = message.document;
|
||||
if (record?.recordKey === lease.recordKey && record.schemaId === lease.schemaId) lease.apply(record);
|
||||
if (record?.recordKey === lease.recordKey && recordMatchesSchema(record, lease.schemaId)) lease.apply(record);
|
||||
}
|
||||
|
||||
private rejectCorrelated(reason: string): void {
|
||||
|
|
@ -628,6 +628,12 @@ export function decodeCultNetPayload<T>(record: CultNetRawDocumentRecord): T {
|
|||
return decode(record.payload) as T;
|
||||
}
|
||||
|
||||
function recordMatchesSchema(record: CultNetRawDocumentRecord, requestedSchema: string): boolean {
|
||||
return record.schemaId === requestedSchema ||
|
||||
record.schemaName === requestedSchema ||
|
||||
record.schemaVersion === requestedSchema;
|
||||
}
|
||||
|
||||
export function decodeCultNetOperationPayload<T>(response: CultNetOperationResponseMessage): T {
|
||||
if (response.payloadEncoding !== "messagepack-base64") {
|
||||
throw new Error(`Unsupported CultNet operation payload '${response.payloadEncoding ?? "unspecified"}'.`);
|
||||
|
|
|
|||
|
|
@ -270,7 +270,10 @@ class FakeProvider {
|
|||
|
||||
private record(): CultNetRawDocumentRecord {
|
||||
return {
|
||||
schemaId: "sample.counter_state.v1",
|
||||
schemaId: "sha256:sample-counter-state",
|
||||
schemaName: "sample.counter_state",
|
||||
schemaVersion: "sample.counter_state.v1",
|
||||
schemaContentHash: "sha256:sample-counter-state-content",
|
||||
recordKey: "counter:main",
|
||||
storedAt: "2026-08-17T00:00:00Z",
|
||||
payloadEncoding: "messagepack",
|
||||
|
|
|
|||
|
|
@ -190,6 +190,9 @@ export interface CultNetDocumentDeleteMessage {
|
|||
|
||||
export interface CultNetRawDocumentRecord {
|
||||
schemaId: string;
|
||||
schemaName?: string;
|
||||
schemaVersion?: string;
|
||||
schemaContentHash?: string;
|
||||
recordKey: string;
|
||||
storedAt: string;
|
||||
payloadEncoding: CultNetRawPayloadEncoding;
|
||||
|
|
@ -1104,6 +1107,9 @@ function validateRawDocumentRecord(
|
|||
|
||||
const candidate = input as Record<string, unknown>;
|
||||
requireNonEmptyString(candidate.schemaId, `${instancePath}/schemaId`);
|
||||
requireOptionalNonEmptyString(candidate.schemaName, `${instancePath}/schemaName`);
|
||||
requireOptionalNonEmptyString(candidate.schemaVersion, `${instancePath}/schemaVersion`);
|
||||
requireOptionalNonEmptyString(candidate.schemaContentHash, `${instancePath}/schemaContentHash`);
|
||||
requireNonEmptyString(candidate.recordKey, `${instancePath}/recordKey`);
|
||||
requireNonEmptyString(candidate.storedAt, `${instancePath}/storedAt`);
|
||||
if (candidate.payloadEncoding !== "messagepack") {
|
||||
|
|
|
|||
|
|
@ -224,6 +224,9 @@ namespace GameCult.Networking
|
|||
Document = new CultNetRawDocumentRecord
|
||||
{
|
||||
SchemaId = binding.SchemaId,
|
||||
SchemaName = descriptor.SchemaName,
|
||||
SchemaVersion = descriptor.SchemaVersion,
|
||||
SchemaContentHash = descriptor.ContentHash,
|
||||
RecordKey = handle.Key.Value,
|
||||
StoredAt = storedAt,
|
||||
PayloadEncoding = "messagepack",
|
||||
|
|
@ -292,6 +295,9 @@ namespace GameCult.Networking
|
|||
documents.Add(new CultNetRawDocumentRecord
|
||||
{
|
||||
SchemaId = binding.SchemaId,
|
||||
SchemaName = descriptor.SchemaName,
|
||||
SchemaVersion = descriptor.SchemaVersion,
|
||||
SchemaContentHash = descriptor.ContentHash,
|
||||
RecordKey = key,
|
||||
StoredAt = storedAt,
|
||||
PayloadEncoding = "messagepack",
|
||||
|
|
|
|||
|
|
@ -608,6 +608,18 @@ namespace GameCult.Networking
|
|||
/// </summary>
|
||||
[Key("schemaId")] public string SchemaId { get; set; } = string.Empty;
|
||||
/// <summary>
|
||||
/// Gets or sets the stable schema family name when the publisher has a typed descriptor.
|
||||
/// </summary>
|
||||
[Key("schemaName")] public string? SchemaName { get; set; }
|
||||
/// <summary>
|
||||
/// Gets or sets the portable versioned schema contract when the publisher has a typed descriptor.
|
||||
/// </summary>
|
||||
[Key("schemaVersion")] public string? SchemaVersion { get; set; }
|
||||
/// <summary>
|
||||
/// Gets or sets the canonical schema content hash when the publisher has a typed descriptor.
|
||||
/// </summary>
|
||||
[Key("schemaContentHash")] public string? SchemaContentHash { get; set; }
|
||||
/// <summary>
|
||||
/// Gets or sets the record key.
|
||||
/// </summary>
|
||||
[Key("recordKey")] public string RecordKey { get; set; } = string.Empty;
|
||||
|
|
|
|||
|
|
@ -1982,6 +1982,9 @@ namespace GameCult.Networking.Tests
|
|||
|
||||
Assert.That(message, Is.Not.Null);
|
||||
Assert.That(message!.Document!.SchemaId, Is.EqualTo(descriptor.SchemaId));
|
||||
Assert.That(message.Document.SchemaName, Is.EqualTo(descriptor.SchemaName));
|
||||
Assert.That(message.Document.SchemaVersion, Is.EqualTo(descriptor.SchemaVersion));
|
||||
Assert.That(message.Document.SchemaContentHash, Is.EqualTo(descriptor.ContentHash));
|
||||
}
|
||||
|
||||
[Test]
|
||||
|
|
|
|||
Loading…
Reference in a new issue