Skip to content

Canonicalize offline sync row footprints #62

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
rdlabo merged 1 commit into from
Aug 12, 2026
Merged

Canonicalize offline sync row footprints #62

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
126 changes: 124 additions & 2 deletions projects/kit/offline/src/lib/offline-sync.service.spec.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -12,7 +12,15 @@ import { OFFLINE_KIT_OPTIONS, type OfflineKitOptions } from './offline-kit-optio
import { OfflineNetworkService } from './offline-network.service';
import { OfflineReplicaPullService } from './offline-replica-pull.service';
import { OfflineReplicaMutationCoordinator } from './offline-replica-mutation-coordinator';
import { defineOfflineReplicaSchema, defineReplicaEntity, integer, naturalKey, generatedId, text } from './offline-replica-schema';
import {
defineOfflineReplicaSchema,
defineReplicaEntity,
integer,
localOnly,
naturalKey,
generatedId,
text,
} from './offline-replica-schema';
import {
canonicalOfflineReplicaIdentity,
OFFLINE_REPOSITORY,
Expand Down Expand Up @@ -43,6 +51,13 @@ const replicaSchema = defineOfflineReplicaSchema({
title: text(),
},
}),
defineReplicaEntity<{ title: string }>()({
table: 'document_views',
sourceKey: 'document_views',
scope: 'partition',
identity: localOnly(),
fields: { title: text() },
}),
],
migrations: [],
});
Expand Down Expand Up @@ -3404,6 +3419,13 @@ describe('OfflineSyncService', () => {
name: text(),
},
}),
defineReplicaEntity<{ title: string }>()({
table: 'test_views',
sourceKey: 'test_views',
scope: 'user',
identity: localOnly(),
fields: { title: text() },
}),
],
migrations: [],
});
Expand All @@ -3414,7 +3436,7 @@ describe('OfflineSyncService', () => {
{ userId: 1, scopeId: '11' },
] as OfflineScope[],
};
const userScopedSourceKeys = new Set(['test_items']);
const userScopedSourceKeys = new Set(['test_items', 'test_views']);

function compareCommands(left: OfflineCommand, right: OfflineCommand): number {
return left.createdAt - right.createdAt || (left.commandId < right.commandId ? -1 : left.commandId > right.commandId ? 1 : 0);
Expand Down Expand Up @@ -3679,6 +3701,106 @@ describe('OfflineSyncService', () => {
});
});

it('multi-scope batchは同じuser-scoped companionのscope alias重複をcommit前に拒否する', async () => {
const repository = TestBed.inject(OFFLINE_REPOSITORY);
const transactReplica = vi.mocked(repository.transactReplica);
const companion = (scopeId: string, title: string): OfflineReplicaRow => ({
userId: 1,
scopeId,
sourceKey: 'test_views',
identity: { kind: 'local', localId: 'shared-view' },
values: { title },
confirmedValues: { title: 'Baseline' },
serverRevision: null,
fetchedAt: 1,
syncState: 'pending',
});

await expect(
service.enqueuePreparedBatch(
async () => [
{
request: {
scopeId: '10',
aggregateType: 'test_items',
identity: { kind: 'generated', localId: 'batch-scope-10' },
operation: 'test_items.create',
payload: {},
optimisticValue: { id: 0, title: 'A' },
},
replicaTransaction: { putRows: [companion('10', 'A')] },
},
{
request: {
scopeId: '11',
aggregateType: 'test_items',
identity: { kind: 'generated', localId: 'batch-scope-11' },
operation: 'test_items.create',
payload: {},
optimisticValue: { id: 0, title: 'B' },
},
replicaTransaction: { putRows: [companion('11', 'B')] },
},
],
{ flush: false },
),
).rejects.toThrow('overlapping replica footprints');

expect(transactReplica).not.toHaveBeenCalled();
expect(commands).toEqual([]);
expect(rows).toHaveLength(1);
});

it('別aggregateの後続enqueueもuser-scoped companionのscope alias共有を拒否する', async () => {
const companion = (scopeId: string, title: string): OfflineReplicaRow => ({
userId: 1,
scopeId,
sourceKey: 'test_views',
identity: { kind: 'local', localId: 'shared-view' },
values: { title },
confirmedValues: { title: 'Baseline' },
serverRevision: null,
fetchedAt: 1,
syncState: 'pending',
});
const firstId = await service.enqueuePrepared(
async () => ({
request: {
scopeId: '10',
aggregateType: 'test_items',
identity: { kind: 'generated', localId: 'scope-10-item' },
operation: 'test_items.create',
payload: {},
optimisticValue: { id: 0, title: 'A' },
},
replicaTransaction: { putRows: [companion('10', 'A')] },
}),
{ flush: false },
);

await expect(
service.enqueuePrepared(
async () => ({
request: {
scopeId: '11',
aggregateType: 'test_items',
identity: { kind: 'generated', localId: 'scope-11-item' },
operation: 'test_items.create',
payload: {},
optimisticValue: { id: 0, title: 'B' },
},
replicaTransaction: { putRows: [companion('11', 'B')] },
}),
{ flush: false },
),
).rejects.toThrow('different aggregates cannot share a replica footprint');

expect(commands).toHaveLength(1);
await service.discard(firstId, { flush: false });
expect(commands).toEqual([]);
expect(findReplicaRow({ userId: 1, scopeId: '11' }, 'test_views', { kind: 'local', localId: 'shared-view' })).toBeUndefined();
});

it('partition-scopedの同一localIdはpartitionごとに独立aggregateのまま並列送信する', async () => {
let resolveFirst!: (value: OfflineCommandResult) => void;
execute.mockImplementationOnce(() => new Promise((resolve) => (resolveFirst = resolve)));
Expand Down
42 changes: 29 additions & 13 deletions projects/kit/offline/src/lib/offline-sync.service.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -23,10 +23,9 @@ import type {
OfflineRepository,
OfflineScope,
} from './offline-repository';
import { OFFLINE_REPOSITORY } from './offline-repository';
import { canonicalOfflineReplicaRowKey, OFFLINE_REPOSITORY } from './offline-repository';
import {
canonicalOfflinePrincipalId,
canonicalOfflineReplicaIdentity,
canonicalOfflineCommandIdentity,
commandIdentityFromReplicaIdentity,
commandIdentityMatchesReplicaRow,
Expand Down Expand Up @@ -350,7 +349,10 @@ export class OfflineSyncService {
const session = await this.#beginEnqueueSession(generation);
this.#assertEnqueueScope(session, request.scopeId);
const materialization = await this.#materializeEnqueue(session.userId, request, replicaTransaction, replaced);
await this.#assertOutboxCapacity(session.userId, [materialization.command], replaced?.commandId);
const currentCommands = await this.#commandsForUser(session.userId);
const retainedCommands = replaced ? currentCommands.filter((command) => command.commandId !== replaced.commandId) : currentCommands;
this.#assertDistinctBatchFootprints([materialization], retainedCommands);
await this.#assertOutboxCapacity(session.userId, [materialization.command], replaced?.commandId, currentCommands);
await this.#commitMaterializedEnqueues([materialization], generation, options, replaced ? [replaced.commandId] : undefined);
return materialization.command.commandId;
}
Expand All @@ -372,16 +374,10 @@ export class OfflineSyncService {
for (const [index, entry] of prepared.entries()) {
this.#assertEnqueueScope(session, entry.request.scopeId);
materializations.push(
await this.#materializeEnqueue(
session.userId,
entry.request,
entry.replicaTransaction,
undefined,
firstCreatedAt + index,
),
await this.#materializeEnqueue(session.userId, entry.request, entry.replicaTransaction, undefined, firstCreatedAt + index),
);
}
this.#assertDistinctBatchFootprints(materializations);
this.#assertDistinctBatchFootprints(materializations, currentCommands);
await this.#assertOutboxCapacity(
session.userId,
materializations.map((item) => item.command),
Expand Down Expand Up @@ -542,9 +538,14 @@ export class OfflineSyncService {
return { command, optimisticRow, optimisticCompanions };
}

#assertDistinctBatchFootprints(entries: readonly MaterializedOfflineEnqueue[]): void {
#assertDistinctBatchFootprints(entries: readonly MaterializedOfflineEnqueue[], existingCommands: readonly OfflineCommand[]): void {
const aggregates = new Set<string>();
const replicaKeys = new Set<string>();
const existingFootprints = new Map<string, string>();
for (const command of existingCommands) {
const aggregate = this.#aggregateKey(command);
for (const key of this.#commandFootprintKeys(command)) existingFootprints.set(key, aggregate);
}
for (const entry of entries) {
const aggregate = this.#aggregateKey(entry.command);
if (aggregates.has(aggregate)) {
Expand All @@ -558,11 +559,26 @@ export class OfflineSyncService {
if (replicaKeys.has(key)) {
throw new Error('Prepared offline batch contains overlapping replica footprints.');
}
const existingAggregate = existingFootprints.get(key);
if (existingAggregate !== undefined && existingAggregate !== aggregate) {
throw new Error('Offline commands for different aggregates cannot share a replica footprint.');
}
replicaKeys.add(key);
}
}
}

#commandFootprintKeys(command: OfflineCommand): readonly string[] {
const identity =
command.identity.kind === 'generated'
? offlineGeneratedReplicaIdentity(command.identity.localId, null)
: ({ kind: 'natural', naturalKey: command.identity.naturalKey } as const);
return [
this.#replicaRowKey({ ...command, identity }),
...(command.optimisticCompanions ?? []).map((companion) => this.#replicaRowKey(companion.key)),
];
}

async #commitMaterializedEnqueues(
entries: readonly MaterializedOfflineEnqueue[],
generation: number,
Expand Down Expand Up @@ -678,7 +694,7 @@ export class OfflineSyncService {
}

#replicaRowKey(key: OfflineReplicaRowKey): string {
return `${canonicalOfflinePrincipalId(key.userId)}:${key.scopeId}:${key.sourceKey}:${canonicalOfflineReplicaIdentity(key.identity)}`;
return canonicalOfflineReplicaRowKey(this.#entitySchema(key.sourceKey), key);
}

async #assertOutboxCapacity(
Expand Down