fix(server): online and storage statistics (#14792)

#### PR Dependency Tree


* **PR #14792** 👈

This tree was auto-generated by
[Charcoal](https://github.com/danerwilliams/charcoal)

<!-- This is an auto-generated comment: release notes by coderabbit.ai
-->
## Summary by CodeRabbit

* **New Features**
* Admin dashboard returns more accurate sync and storage timelines with
carry‑forwarded minute buckets and corrected current totals.

* **Bug Fixes**
* Active-user flushes are debounced/scheduled to prevent overlapping
writes and reduce stale counts.
* Snapshot writes now retry and will skip gracefully when lock
contention prevents completion, avoiding partial snapshots.

* **Tests**
* New e2e tests cover carry‑forward behavior, no backfill outside
requested windows, and storage history accuracy.
<!-- end of auto-generated comment: release notes by coderabbit.ai -->

packages/backend/server/src/core/sync/gateway.ts

@@ -218,6 +218,9 @@ export class SpaceSyncGateway
   private readonly localUserConnectionCounts = new Map<string, number>();
   private unresolvedPresenceSockets = 0;
   private flushTimer?: NodeJS.Timeout;
+  private activeUsersFlushTimer?: NodeJS.Timeout;
+  private activeUsersFlushInFlight = false;
+  private activeUsersFlushQueued = false;
 
   constructor(
     private readonly ac: AccessController,
@@ -229,12 +232,9 @@ export class SpaceSyncGateway
   ) {}
 
   onModuleInit() {
+    this.scheduleActiveUsersFlush(0);
     this.flushTimer = setInterval(() => {
-      this.flushActiveUsersMinute().catch(error => {
-        this.logger.warn(
-          `Failed to flush active users minute: ${this.formatError(error)}`
-        );
-      });
+      this.scheduleActiveUsersFlush(0);
     }, 60_000);
     this.flushTimer.unref?.();
   }
@@ -244,6 +244,11 @@ export class SpaceSyncGateway
       clearInterval(this.flushTimer);
       this.flushTimer = undefined;
     }
+    if (this.activeUsersFlushTimer) {
+      clearTimeout(this.activeUsersFlushTimer);
+      this.activeUsersFlushTimer = undefined;
+    }
+    this.activeUsersFlushQueued = false;
   }
 
   private encodeUpdates(updates: Uint8Array[]) {
@@ -331,13 +336,7 @@ export class SpaceSyncGateway
     metrics.socketio.gauge('connections').record(this.connectionCount);
     const userId = this.attachPresenceUserId(client);
     this.trackConnectedSocket(client.id, userId);
-    void this.flushActiveUsersMinute({
-      aggregateAcrossCluster: false,
-    }).catch(error => {
-      this.logger.warn(
-        `Failed to flush active users minute: ${this.formatError(error)}`
-      );
-    });
+    this.scheduleActiveUsersFlush();
   }
 
   handleDisconnect(client: Socket) {
@@ -347,13 +346,7 @@ export class SpaceSyncGateway
       `Connection disconnected, total: ${this.connectionCount}`
     );
     metrics.socketio.gauge('connections').record(this.connectionCount);
-    void this.flushActiveUsersMinute({
-      aggregateAcrossCluster: false,
-    }).catch(error => {
-      this.logger.warn(
-        `Failed to flush active users minute: ${this.formatError(error)}`
-      );
-    });
+    this.scheduleActiveUsersFlush();
   }
 
   private attachPresenceUserId(client: Socket): string | null {
@@ -435,13 +428,55 @@ export class SpaceSyncGateway
     }
   }
 
+  private scheduleActiveUsersFlush(delayMs = 250) {
+    if (this.activeUsersFlushTimer) {
+      return;
+    }
+
+    if (this.activeUsersFlushInFlight) {
+      this.activeUsersFlushQueued = true;
+      return;
+    }
+
+    this.activeUsersFlushTimer = setTimeout(() => {
+      this.activeUsersFlushTimer = undefined;
+      this.runScheduledActiveUsersFlush();
+    }, delayMs);
+    this.activeUsersFlushTimer.unref?.();
+  }
+
+  private runScheduledActiveUsersFlush() {
+    if (this.activeUsersFlushInFlight) {
+      this.activeUsersFlushQueued = true;
+      return;
+    }
+
+    this.activeUsersFlushInFlight = true;
+    void this.flushActiveUsersMinute()
+      .catch(error => {
+        this.logger.warn(
+          `Failed to flush active users minute: ${this.formatError(error)}`
+        );
+      })
+      .finally(() => {
+        this.activeUsersFlushInFlight = false;
+        if (this.activeUsersFlushQueued) {
+          this.activeUsersFlushQueued = false;
+          this.scheduleActiveUsersFlush(0);
+        }
+      });
+  }
+
   private async flushActiveUsersMinute(options?: {
     aggregateAcrossCluster?: boolean;
+    skipWriteOnAggregateError?: boolean;
   }) {
     const minute = new Date();
     minute.setSeconds(0, 0);
 
     const aggregateAcrossCluster = options?.aggregateAcrossCluster ?? true;
+    const skipWriteOnAggregateError =
+      options?.skipWriteOnAggregateError ?? aggregateAcrossCluster;
     let activeUsers = this.resolveLocalActiveUsers();
     if (aggregateAcrossCluster) {
       try {
@@ -467,8 +502,9 @@ export class SpaceSyncGateway
         }
       } catch (error) {
         this.logger.warn(
-          `Failed to aggregate active users from sockets, using local value: ${this.formatError(error)}`
+          `Failed to aggregate active users from sockets: ${this.formatError(error)}`
         );
+        if (skipWriteOnAggregateError) return;
       }
     }
 

packages/backend/server/src/core/workspaces/stats.job.ts

@@ -5,9 +5,11 @@ import { Prisma, PrismaClient } from '@prisma/client';
 import { metrics } from '../../base';
 
 const LOCK_NAMESPACE = 97_301;
-const LOCK_KEY = 1;
+const REFRESH_LOCK_KEY = 1;
 const DIRTY_BATCH_SIZE = 500;
 const FULL_REFRESH_BATCH_SIZE = 2000;
+const REFRESH_LOCK_RETRY_DELAY_MS = 5_000;
+const REFRESH_LOCK_RETRY_TIMES = 12;
 const TRANSACTION_TIMEOUT_MS = 120_000;
 
 @Injectable()
@@ -21,7 +23,7 @@ export class WorkspaceStatsJob {
     const started = Date.now();
 
     try {
-      const result = await this.withAdvisoryLock(async tx => {
+      const result = await this.withAdvisoryLock(REFRESH_LOCK_KEY, async tx => {
         const backlog = await this.countDirty(tx);
         metrics.workspace
           .gauge('admin_stats_dirty_backlog')
@@ -63,11 +65,12 @@ export class WorkspaceStatsJob {
   async recalibrate() {
     let lastSid = 0;
     let processed = 0;
+    let completed = true;
 
     while (true) {
       const started = Date.now();
       try {
-        const result = await this.withAdvisoryLock(async tx => {
+        const result = await this.withRefreshLockRetry(async tx => {
           const workspaces = await this.fetchWorkspaceBatch(
             tx,
             lastSid,
@@ -87,8 +90,9 @@ export class WorkspaceStatsJob {
         });
 
         if (!result) {
-          this.logger.debug(
-            'skip admin stats recalibration, lock not acquired'
+          completed = false;
+          this.logger.warn(
+            'skip admin stats recalibration after retrying lock acquisition'
           );
           break;
         }
@@ -108,6 +112,7 @@ export class WorkspaceStatsJob {
           break;
         }
       } catch (error) {
+        completed = false;
         metrics.workspace.counter('admin_stats_refresh_failed').add(1, {
           mode: 'full',
         });
@@ -125,13 +130,24 @@ export class WorkspaceStatsJob {
       );
     }
 
+    if (!completed) {
+      this.logger.warn(
+        'Skip daily workspace admin stats snapshot because full recalibration did not complete'
+      );
+      return;
+    }
+
     try {
-      const snapshotted = await this.withAdvisoryLock(async tx => {
+      const snapshotted = await this.withRefreshLockRetry(async tx => {
         await this.writeDailySnapshot(tx);
         return true;
       });
       if (snapshotted) {
         this.logger.debug('Wrote daily workspace admin stats snapshot');
+      } else {
+        this.logger.warn(
+          'Skipped daily workspace admin stats snapshot after retrying lock acquisition'
+        );
       }
     } catch (error) {
       this.logger.error(
@@ -142,9 +158,10 @@ export class WorkspaceStatsJob {
   }
 
   private async withAdvisoryLock<T>(
+    lockKey: number,
     callback: (tx: Prisma.TransactionClient) => Promise<T>
   ): Promise<T | null> {
-    const lockIdSql = Prisma.sql`(${LOCK_NAMESPACE}::bigint << 32) + ${LOCK_KEY}::bigint`;
+    const lockIdSql = Prisma.sql`(${LOCK_NAMESPACE}::bigint << 32) + ${lockKey}::bigint`;
 
     return await this.prisma.$transaction(
       async tx => {
@@ -169,6 +186,26 @@ export class WorkspaceStatsJob {
     );
   }
 
+  private async withRefreshLockRetry<T>(
+    callback: (tx: Prisma.TransactionClient) => Promise<T>
+  ) {
+    for (let attempt = 0; attempt < REFRESH_LOCK_RETRY_TIMES; attempt++) {
+      const result = await this.withAdvisoryLock(REFRESH_LOCK_KEY, callback);
+
+      if (result) {
+        return result;
+      }
+
+      if (attempt < REFRESH_LOCK_RETRY_TIMES - 1) {
+        await new Promise(resolve =>
+          setTimeout(resolve, REFRESH_LOCK_RETRY_DELAY_MS)
+        );
+      }
+    }
+
+    return null;
+  }
+
   private async loadDirty(
     tx: Prisma.TransactionClient,
     limit: number