fix(kit): harden offline auth revocation and recovery

projects/kit/offline/src/lib/offline-coordinator.service.spec.ts

@@ -22,13 +22,16 @@ describe('OfflineCoordinatorService', () => {
     };
     const session = {
       initialize: vi.fn(async () => undefined),
-      activateSession: vi.fn(async (userId: number, _scopeIds: readonly number[], _authSubject: string | null, lease: { isCurrent(): boolean }) => {
-        order.push('activate-remote');
-        if (!lease.isCurrent()) return false;
-        sessionState.userId = userId;
-        return true;
-      }),
+      activateSession: vi.fn(
+        async (userId: number, _scopeIds: readonly number[], _authSubject: string | null, lease: { isCurrent(): boolean }) => {
+          order.push('activate-remote');
+          if (!lease.isCurrent()) return false;
+          sessionState.userId = userId;
+          return true;
+        },
+      ),
       suspendRemoteSession: vi.fn(async () => void order.push('suspend-remote')),
+      revokeAccess: vi.fn(() => void order.push('revoke')),
       activateOfflineSession: vi.fn(async () => {
         order.push('activate-local');
         return manifest;
@@ -44,6 +47,7 @@ describe('OfflineCoordinatorService', () => {
       conflicts: signal([]),
       initialize: vi.fn(async () => undefined),
       resetSession: vi.fn(async () => void order.push('reset')),
+      revokeSession: vi.fn(() => void order.push('revoke-sync')),
       refreshSession: vi.fn(async () => void order.push('resume-remote')),
       refreshLocalSession: vi.fn(async () => void order.push('refresh-local')),
       discardAllPending: vi.fn(async () => undefined),
@@ -119,6 +123,16 @@ describe('OfflineCoordinatorService', () => {
     expect(sessionState.userId).toBeNull();
   });
 
+  it('revokes runtime local access synchronously before queued durable cleanup', async () => {
+    const { coordinator, session, sync } = setup();
+
+    const clearing = coordinator.clearActiveSession();
+
+    expect(session.revokeAccess).toHaveBeenCalledOnce();
+    expect(sync.revokeSession).toHaveBeenCalledOnce();
+    await clearing;
+  });
+
   it('keeps a newer identity when an older activation completes late', async () => {
     const { coordinator, session, sessionState } = setup();
     let releaseOld: (() => void) | undefined;

projects/kit/offline/src/lib/offline-coordinator.service.ts

@@ -59,10 +59,7 @@ export class OfflineCoordinatorService {
   /**
    * Activates a restored identity for local replica/outbox use without enabling transport sync.
    */
-  activateOfflineSession(
-    authSubject?: string | null,
-    authLease?: OfflineSessionTransitionLease,
-  ): Promise<OfflineSessionManifest | null> {
+  activateOfflineSession(authSubject?: string | null, authLease?: OfflineSessionTransitionLease): Promise<OfflineSessionManifest | null> {
     const revision = ++this.#transitionRevision;
     const lease = this.#lease(revision, authLease);
     return this.#enqueueTransition(async () => {
@@ -75,6 +72,8 @@ export class OfflineCoordinatorService {
   }
 
   clearActiveSession(): Promise<void> {
+    this.#sync.revokeSession();
+    this.#session.revokeAccess();
     ++this.#transitionRevision;
     return this.#enqueueTransition(async () => {
       await this.#sync.resetSession();

projects/kit/offline/src/lib/offline-session.service.spec.ts

@@ -60,6 +60,16 @@ describe('OfflineSessionService shared-device boundary', () => {
     await expect(service.getSession()).resolves.toBeNull();
   });
 
+  it('永続削除を待たずにlocal/outboxとremote syncのruntime accessを失効する', async () => {
+    await service.activateOfflineSession('uid-A');
+
+    service.revokeAccess();
+
+    await expect(service.getLocalSession()).resolves.toBeNull();
+    await expect(service.getSession()).resolves.toBeNull();
+    await expect(service.getOfflineAccessManifest('uid-A')).resolves.toMatchObject({ userId: 10 });
+  });
+
   it('既知のsubjectがmanifestと違う場合はlocal accessを拒否する', async () => {
     await expect(service.getOfflineAccessManifest('uid-B')).resolves.toBeNull();
     await expect(service.getOfflineAccessManifest(null)).resolves.toBeNull();

projects/kit/offline/src/lib/offline-session.service.ts

@@ -98,6 +98,12 @@ export class OfflineSessionService {
     this.#remoteActivatedThisRun = false;
   }
 
+  /** Immediately revoke local and remote runtime access while durable cleanup is pending. */
+  revokeAccess(): void {
+    this.#localAccessThisRun = false;
+    this.#remoteActivatedThisRun = false;
+  }
+
   /** Disable remote pull/replay eligibility while retaining the verified local manifest. */
   async suspendRemoteSession(): Promise<void> {
     await this.initialize();
@@ -134,10 +140,7 @@ export class OfflineSessionService {
    * @param authSubject - A currently known provider subject. When supplied, it must match the
    * persisted subject.
    */
-  async activateOfflineSession(
-    authSubject?: string | null,
-    lease?: OfflineSessionTransitionLease,
-  ): Promise<OfflineSessionManifest | null> {
+  async activateOfflineSession(authSubject?: string | null, lease?: OfflineSessionTransitionLease): Promise<OfflineSessionManifest | null> {
     const manifest = await this.getOfflineAccessManifest(authSubject);
     if (lease && !lease.isCurrent()) return null;
     this.#localAccessThisRun = manifest !== null;

projects/kit/offline/src/lib/offline-sync.service.spec.ts

@@ -44,6 +44,7 @@ describe('OfflineSyncService', () => {
   let session: { userId: number; scopes: OfflineScope[] } | null;
   let localSession: { userId: number; scopes: OfflineScope[] } | null | undefined;
   let beforePutCommand: ((command: OfflineCommand) => Promise<void>) | null;
+  let beforeGetReplicaRow: (() => Promise<void>) | null;
   let pull: ReturnType<typeof vi.fn<(scope: OfflineScope) => Promise<void>>>;
   let handleError: ReturnType<typeof vi.fn<(error: unknown) => void>>;
   const execute = vi.fn(
@@ -57,6 +58,7 @@ describe('OfflineSyncService', () => {
     session = { userId: 1, scopes: [{ userId: 1, groupId: 10 }] };
     localSession = undefined;
     beforePutCommand = null;
+    beforeGetReplicaRow = null;
     pull = vi.fn(async () => undefined);
     handleError = vi.fn();
     execute.mockReset();
@@ -81,13 +83,15 @@ describe('OfflineSyncService', () => {
       removeCommand: vi.fn(async (commandId: string) => {
         commands = commands.filter((item) => item.commandId !== commandId);
       }),
-      getReplicaRow: vi.fn(
-        async (scope: OfflineScope, sourceKey: string, localId: string) =>
+      getReplicaRow: vi.fn(async (scope: OfflineScope, sourceKey: string, localId: string) => {
+        await beforeGetReplicaRow?.();
+        return (
           rows.find(
             (item) =>
               item.userId === scope.userId && item.groupId === scope.groupId && item.sourceKey === sourceKey && item.localId === localId,
-          ) ?? null,
-      ),
+          ) ?? null
+        );
+      }),
       getReplicaRowByServerId: vi.fn(
         async (scope: OfflineScope, sourceKey: string, serverId: number) =>
           rows.find(
@@ -172,6 +176,65 @@ describe('OfflineSyncService', () => {
     expect(service.pendingCount()).toBe(1);
   });
 
+  it('session失効前に開始したenqueueを永続commitせずreset完了まで直列化する', async () => {
+    let releaseRead: (() => void) | undefined;
+    let readStarted: (() => void) | undefined;
+    const started = new Promise<void>((resolve) => {
+      readStarted = resolve;
+    });
+    const gate = new Promise<void>((resolve) => {
+      releaseRead = resolve;
+    });
+    beforeGetReplicaRow = async () => {
+      readStarted?.();
+      await gate;
+    };
+
+    const enqueue = service.enqueue(
+      {
+        groupId: 10,
+        aggregateType: 'documents',
+        aggregateLocalId: 'revoked',
+        operation: 'documents.create',
+        payload: { title: 'stale' },
+        optimisticValue: { id: 0, title: 'stale' },
+      },
+      { flush: false },
+    );
+    await started;
+
+    service.revokeSession();
+    const reset = service.resetSession();
+    releaseRead?.();
+
+    await expect(enqueue).rejects.toThrow('Offline session changed');
+    await reset;
+    expect(rows).toEqual([]);
+    expect(commands).toEqual([]);
+  });
+
+  it('旧flushが失敗してもresetを中断せずdurable cleanupへ進める', async () => {
+    const pullError = new Error('pull failed during revocation');
+    let rejectPull: ((error: unknown) => void) | undefined;
+    pull.mockImplementationOnce(
+      () =>
+        new Promise<void>((_resolve, reject) => {
+          rejectPull = reject;
+        }),
+    );
+    connected.set(true);
+    const flush = service.flush();
+    const flushRejected = expect(flush).rejects.toBe(pullError);
+    await vi.waitFor(() => expect(pull).toHaveBeenCalledOnce());
+
+    const reset = service.resetSession();
+    rejectPull?.(pullError);
+
+    await flushRejected;
+    await expect(reset).resolves.toBeUndefined();
+    expect(service.pendingCount()).toBe(0);
+  });
+
   it('同じaggregateの操作を作成順に送り、成功後だけoutboxから除く', async () => {
     await service.enqueue(
       {
@@ -505,12 +568,13 @@ describe('OfflineSyncService', () => {
     connected.set(true);
     const oldFlush = service.flush();
     await vi.waitFor(() => expect(execute).toHaveBeenCalledOnce());
-    await service.resetSession();
+    const reset = service.resetSession();
+    resolveExecute({ response: null, serverRevision: 2 });
+    await reset;
     commands = commands.filter((command) => command.userId !== 1);
     connected.set(false);
     session = { userId: 2, scopes: [{ userId: 2, groupId: 20 }] };
     await service.refreshSession();
-    resolveExecute({ response: null, serverRevision: 2 });
     await oldFlush;
     expect(execute).toHaveBeenCalledOnce();
     expect(commands.some((command) => command.userId === 1)).toBe(false);
@@ -534,10 +598,11 @@ describe('OfflineSyncService', () => {
     const oldFlush = service.flush();
     await vi.waitFor(() => expect(execute).toHaveBeenCalledOnce());
     connected.set(false);
-    await service.resetSession();
+    const reset = service.resetSession();
+    resolveFirst({ response: null });
+    await reset;
     await service.refreshSession();
     expect(service.pendingCommands()[0]?.state).toBe('pending');
-    resolveFirst({ response: null });
     await oldFlush;
     connected.set(true);
     await service.flush();

projects/kit/offline/src/lib/offline-sync.service.ts

@@ -56,6 +56,7 @@ export class OfflineSyncService {
   readonly #knownScopes = new Map<string, OfflineScope>();
   #activeUserId: number | null = null;
   #flushPromise: Promise<void> | null = null;
+  readonly #flushTransitions = new Set<Promise<void>>();
   #generation = 0;
   readonly #sendingTransitions = new Set<Promise<void>>();
   #enqueueTail: Promise<void> = Promise.resolve();
@@ -120,7 +121,8 @@ export class OfflineSyncService {
   }
 
   async resetSession(): Promise<void> {
-    this.#invalidateFlush();
+    this.revokeSession();
+    await Promise.allSettled([this.#enqueueTail, ...this.#flushTransitions]);
     await this.#waitForSendingTransitions();
     await this.#restoreInterruptedCommands();
     this.#activeUserId = null;
@@ -129,16 +131,22 @@ export class OfflineSyncService {
     this.#scheduleRetry(null);
   }
 
+  /** Synchronously invalidate in-flight enqueue and transport work owned by the current session. */
+  revokeSession(): void {
+    this.#invalidateFlush();
+  }
+
   enqueue<T>(request: EnqueueOfflineCommand<T>, options: { flush?: boolean } = {}): Promise<string> {
-    const enqueue = this.#enqueueTail.then(() => this.#enqueue(request, options));
+    const generation = this.#generation;
+    const enqueue = this.#enqueueTail.then(() => this.#enqueue(request, options, generation));
     this.#enqueueTail = enqueue.then(
       () => undefined,
       () => undefined,
     );
     return enqueue;
   }
 
-  async #enqueue<T>(request: EnqueueOfflineCommand<T>, options: { flush?: boolean }): Promise<string> {
+  async #enqueue<T>(request: EnqueueOfflineCommand<T>, options: { flush?: boolean }, generation: number): Promise<string> {
     await this.initialize();
     const session = await this.#getLocalSession();
     if (!session) throw new Error('Cannot enqueue an offline command without an authenticated user');
@@ -186,6 +194,9 @@ export class OfflineSyncService {
       fetchedAt: Date.now(),
       syncState: 'pending',
     };
+    if (generation !== this.#generation) {
+      throw new Error('Offline session changed before the command could be persisted');
+    }
     await this.#repository.transactReplica({ putRows: [optimisticRow], putCommands: [command] });
     await this.#refreshState();
     if (options.flush !== false && this.#network.connected()) this.#flushInBackground();
@@ -223,9 +234,11 @@ export class OfflineSyncService {
     if (this.#flushPromise) return this.#flushPromise;
     const generation = this.#generation;
     const promise = this.#runFlush(generation).finally(() => {
+      this.#flushTransitions.delete(promise);
       if (this.#flushPromise === promise) this.#flushPromise = null;
     });
     this.#flushPromise = promise;
+    this.#flushTransitions.add(promise);
     return promise;
   }