chore(supersync): warn when full-state aggregate scan approaches tx timeout

The aggregate now runs inside the 60s upload transaction. Typical histories
complete in <100ms, but pathological cases (millions of ops, long cleanup
retention) could approach the limit. Mirror the existing >5s slow-aggregate
warning from OperationDownloadService so production logs surface the issue
before it becomes a hard timeout.

packages/super-sync-server/src/sync/sync.service.ts

@@ -54,6 +54,11 @@ interface LatestEntityOperationRow {
 // Conservative enough to avoid planner-heavy BitmapOr + Sort plans on large
 // histories while still replacing up to 100 per-entity round trips with one query.
 const CONFLICT_DETECTION_ENTITY_BATCH_SIZE = 100;
+// Observability threshold: log a warning when the full-state op aggregate scan
+// exceeds this duration. Mirrors the threshold used by the legacy snapshot
+// vector-clock aggregate in OperationDownloadService so production logs use a
+// consistent slow-aggregate signal.
+const SLOW_FULL_STATE_AGGREGATE_MS = 5_000;
 const OLD_OPS_CLEANUP_DELETE_BATCH_SIZE = 5_000;
 const OLD_OPS_CLEANUP_MAX_DELETED_PER_RUN = 25_000;
 // Operator-DoS guardrail: a 1M `take:` materializes 1M ids in Node memory and
@@ -587,12 +592,17 @@ export class SyncService {
    * time so the persisted `latest_full_state_vector_clock` reflects every
    * client whose ops may still live in conflict detection — not just the
    * clients named on the snapshot op itself.
+   *
+   * Logs a warning when the scan exceeds `SLOW_FULL_STATE_AGGREGATE_MS` so
+   * pathological histories (millions of ops, cleanup retention too long) are
+   * observable in production before they approach the 60s upload-tx timeout.
    */
   private async _aggregatePriorVectorClock(
     tx: Prisma.TransactionClient,
     userId: number,
     beforeServerSeq: number,
   ): Promise<VectorClock> {
+    const startedAt = Date.now();
     const rows = await tx.$queryRaw<Array<{ client_id: string; max_counter: bigint }>>`
       SELECT kv.key AS client_id, MAX(kv.value::bigint) AS max_counter
       FROM operations, LATERAL jsonb_each_text(vector_clock) AS kv(key, value)
@@ -606,6 +616,14 @@ export class SyncService {
     for (const row of rows) {
       out[row.client_id] = Number(row.max_counter);
     }
+    const elapsedMs = Date.now() - startedAt;
+    if (elapsedMs > SLOW_FULL_STATE_AGGREGATE_MS) {
+      Logger.warn(
+        `[user:${userId}] Full-state op aggregate scan took ${elapsedMs}ms ` +
+          `(${rows.length} clients, beforeSeq=${beforeServerSeq}); approaching ` +
+          `upload-tx timeout. Investigate history size and cleanup retention.`,
+      );
+    }
     return out;
   }