fix(firo): make Spark anon-set sync resumable

Author: reubenyap

lib/db/sqlite/firo_cache_coordinator.dart

@@ -84,72 +84,228 @@ abstract class FiroCacheCoordinator {
     });
   }
 
+  /// Sync the Spark anonymity set cache for `groupId` from the node.
+  ///
+  /// Each sector the server returns is persisted to disk in its own SQLite
+  /// transaction against an in-progress SparkSet row (`complete = 0`). The
+  /// row and its coins are invisible to readers until
+  /// [_markSparkAnonSetComplete] flips the flag after a strict integrity
+  /// check (linked coin count == expected delta) passes.
+  ///
+  /// Resumability: if a prior sync crashed mid-download this function
+  /// picks up at the count of coins already linked to the in-progress
+  /// row, never re-downloading sectors that committed. If the server's
+  /// blockHash has shifted between attempts the partial row is discarded
+  /// (orderKey indices no longer align at the new blockHash) and the
+  /// delta is fetched fresh.
+  ///
+  /// Integrity: the finalize step rolls back and leaves `complete = 0`
+  /// if the link count doesn't match the expected delta — a partial or
+  /// over-full cache never becomes the current set.
   static Future<void> runFetchAndUpdateSparkAnonSetCacheForGroupId(
     int groupId,
     ElectrumXClient client,
     CryptoCurrencyNetwork network,
     void Function(int countFetched, int totalCount)? progressUpdated,
   ) async {
     await _setLocks[network]!.protect(() async {
-      const sectorSize =
-          1500; // chosen as a somewhat decent value. Could be changed in the future if wanted/needed
+      const sectorSize = 1500;
+
       final prevMeta = await FiroCacheCoordinator.getLatestSetInfoForGroupId(
         groupId,
         network,
       );
-
       final prevSize = prevMeta?.size ?? 0;
 
       final meta = await client.getSparkAnonymitySetMeta(coinGroupId: groupId);
 
-      progressUpdated?.call(prevSize, meta.size);
+      void updateProgress(int fetchedDelta) {
+        progressUpdated?.call(prevSize + fetchedDelta, meta.size);
+      }
 
       if (prevMeta?.blockHash == meta.blockHash) {
-        Logging.instance.d("prevMeta?.blockHash == meta.blockHash");
+        updateProgress(0);
+        if (prevMeta?.size == meta.size) {
+          Logging.instance.d(
+            "Spark anon set groupId=$groupId already up to date "
+            "(blockHash=${meta.blockHash}, size=${meta.size})",
+          );
+          return;
+        }
+        // Server reports a different size for the same blockHash. On
+        // Firo's server this should be impossible (blockHash advances
+        // whenever a coin is added), so this is treated as an anomaly.
+        // Refuse to sync: appending coins to the existing finalized row
+        // (what INSERT OR IGNORE in the writer would produce) would leak
+        // unverified coins into a set whose setHash we've already
+        // committed to.
+        Logging.instance.w(
+          "Spark anon set groupId=$groupId server reports different size "
+          "(${prevMeta!.size} -> ${meta.size}) at same blockHash "
+          "${meta.blockHash}; skipping sync to preserve cached state.",
+        );
         return;
       }
 
       final numberOfCoinsToFetch = meta.size - prevSize;
+      if (numberOfCoinsToFetch < 0) {
+        updateProgress(0);
+        // Reorg-style shrink: server has fewer coins than our last
+        // finalized set. Refuse to sync rather than invalidate cached data.
+        Logging.instance.w(
+          "Spark anon set groupId=$groupId appears to have shrunk "
+          "($prevSize -> ${meta.size}); skipping sync to preserve "
+          "cached data.",
+        );
+        return;
+      }
+      if (numberOfCoinsToFetch == 0) {
+        updateProgress(0);
+        // blockHash advanced but no new coins in this group's set. We do
+        // not materialise a new SparkSet row for an empty delta — a
+        // same-size row would create a tiebreaker ambiguity in
+        // _getLatestSetInfoForGroupId. Drop any stray in-progress row so
+        // it doesn't confuse the next resume attempt.
+        final stale = await _Reader._getIncompleteSetForGroupId(
+          groupId,
+          db: _FiroCache.setCacheDB(network),
+        );
+        if (stale.isNotEmpty) {
+          await _workers[network]!.runTask(
+            FCTask(
+              func: FCFuncName._deleteIncompleteSparkSetsForGroup,
+              data: groupId,
+            ),
+          );
+        }
+        return;
+      }
+
+      // Decide whether to resume an existing in-progress row or start
+      // fresh. Cases:
+      //   * no in-progress row                     -> cursor = 0
+      //   * in-progress blockHash differs          -> discard, cursor = 0
+      //   * in-progress linked > expected delta    -> corrupt, discard,
+      //                                               cursor = 0
+      //   * in-progress blockHash matches          -> cursor = linkedSoFar
+      final incomplete = await _Reader._getIncompleteSetForGroupId(
+        groupId,
+        db: _FiroCache.setCacheDB(network),
+      );
 
-      final fullSectorCount = numberOfCoinsToFetch ~/ sectorSize;
-      final remainder = numberOfCoinsToFetch % sectorSize;
+      int cursor;
+      if (incomplete.isEmpty) {
+        cursor = 0;
+      } else if (incomplete.length > 1) {
+        Logging.instance.w(
+          "Spark anon set groupId=$groupId has ${incomplete.length} "
+          "in-progress rows; discarding ambiguous state.",
+        );
+        await _workers[network]!.runTask(
+          FCTask(
+            func: FCFuncName._deleteIncompleteSparkSetsForGroup,
+            data: groupId,
+          ),
+        );
+        cursor = 0;
+      } else {
+        final incBlockHash = incomplete.first["blockHash"] as String;
+        final incSetHash = incomplete.first["setHash"] as String;
+        final incSetId = incomplete.first["id"] as int;
+
+        // Discard the in-progress row if either blockHash or setHash
+        // disagrees with the current meta. blockHash disagreement means
+        // the server's indexing has shifted; setHash disagreement at the
+        // same blockHash would indicate the in-progress row targets a
+        // different set snapshot and resuming would mix coin contexts.
+        if (incBlockHash != meta.blockHash ||
+            incSetHash != meta.setHash) {
+          Logging.instance.i(
+            "Spark anon set groupId=$groupId in-progress "
+            "(blockHash=$incBlockHash, setHash=$incSetHash) does not "
+            "match meta (blockHash=${meta.blockHash}, "
+            "setHash=${meta.setHash}); discarding in-flight row.",
+          );
+          await _workers[network]!.runTask(
+            FCTask(
+              func: FCFuncName._deleteIncompleteSparkSetsForGroup,
+              data: groupId,
+            ),
+          );
+          cursor = 0;
+        } else {
+          final linked = await _Reader._countSetCoins(
+            incSetId,
+            db: _FiroCache.setCacheDB(network),
+          );
+          if (linked > numberOfCoinsToFetch) {
+            Logging.instance.w(
+              "Spark anon set groupId=$groupId in-progress row has "
+              "$linked linked coins but delta is only "
+              "$numberOfCoinsToFetch; discarding in-flight row.",
+            );
+            await _workers[network]!.runTask(
+              FCTask(
+                func: FCFuncName._deleteIncompleteSparkSetsForGroup,
+                data: groupId,
+              ),
+            );
+            cursor = 0;
+          } else {
+            cursor = linked;
+          }
+        }
+      }
 
-      final List<dynamic> coins = [];
+      updateProgress(cursor);
+
+      while (cursor < numberOfCoinsToFetch) {
+        final endIndex = cursor + sectorSize <= numberOfCoinsToFetch
+            ? cursor + sectorSize
+            : numberOfCoinsToFetch;
+        final expected = endIndex - cursor;
 
-      for (int i = 0; i < fullSectorCount; i++) {
-        final start = (i * sectorSize);
         final data = await client.getSparkAnonymitySetBySector(
           coinGroupId: groupId,
           latestBlock: meta.blockHash,
-          startIndex: start,
-          endIndex: start + sectorSize,
+          startIndex: cursor,
+          endIndex: endIndex,
         );
-        progressUpdated?.call(start + sectorSize, numberOfCoinsToFetch);
 
-        coins.addAll(data);
-      }
+        // Refuse to persist a sector whose size doesn't match the request:
+        // a partial or over-full server response would break the finalize-
+        // time integrity check and potentially skew the resume cursor.
+        if (data.length != expected) {
+          throw Exception(
+            "Spark anon set sector size mismatch for groupId=$groupId: "
+            "requested $expected coins in range [$cursor, $endIndex), "
+            "server returned ${data.length}",
+          );
+        }
+
+        final sectorCoins = data
+            .map((e) => RawSparkCoin.fromRPCResponse(e as List, groupId))
+            .toList();
 
-      if (remainder > 0) {
-        final data = await client.getSparkAnonymitySetBySector(
-          coinGroupId: groupId,
-          latestBlock: meta.blockHash,
-          startIndex: numberOfCoinsToFetch - remainder,
-          endIndex: numberOfCoinsToFetch,
+        await _workers[network]!.runTask(
+          FCTask(
+            func: FCFuncName._insertSparkAnonSetCoinsIncremental,
+            data: (meta, sectorCoins, cursor),
+          ),
         );
-        progressUpdated?.call(numberOfCoinsToFetch, numberOfCoinsToFetch);
 
-        coins.addAll(data);
+        cursor = endIndex;
+        updateProgress(cursor);
       }
 
-      final result =
-          coins
-              .map((e) => RawSparkCoin.fromRPCResponse(e as List, groupId))
-              .toList();
-
+      // All sectors persisted. Flip `complete = 1` iff the link count
+      // matches the expected delta. On failure the in-progress row stays
+      // hidden and the error propagates; the next sync will observe the
+      // over-linked row and reset.
       await _workers[network]!.runTask(
         FCTask(
-          func: FCFuncName._updateSparkAnonSetCoinsWith,
-          data: (meta, result),
+          func: FCFuncName._markSparkAnonSetComplete,
+          data: (meta, numberOfCoinsToFetch),
         ),
       );
     });
@@ -250,6 +406,44 @@ abstract class FiroCacheCoordinator {
         .toList();
   }
 
+  static Future<({SparkAnonymitySetMeta meta, List<RawSparkCoin> coins})?>
+      getSetCoinsAndLatestSetInfoForGroupId(
+    int groupId,
+    CryptoCurrencyNetwork network,
+  ) async {
+    final resultSet = await _Reader._getSetCoinsAndLatestSetInfoForGroupId(
+      groupId,
+      db: _FiroCache.setCacheDB(network),
+    );
+    if (resultSet.isEmpty) {
+      return null;
+    }
+
+    final first = resultSet.first;
+    final coins = resultSet
+        .map(
+          (row) => RawSparkCoin(
+            serialized: row["serialized"] as String,
+            txHash: row["txHash"] as String,
+            context: row["context"] as String,
+            groupId: groupId,
+          ),
+        )
+        .toList()
+        .reversed
+        .toList();
+
+    return (
+      meta: SparkAnonymitySetMeta(
+        coinGroupId: groupId,
+        blockHash: first["blockHash"] as String,
+        setHash: first["setHash"] as String,
+        size: first["size"] as int,
+      ),
+      coins: coins,
+    );
+  }
+
   static Future<SparkAnonymitySetMeta?> getLatestSetInfoForGroupId(
     int groupId,
     CryptoCurrencyNetwork network,