diff --git a/docs/SYNC_SKIPPED_ENTITY_PERSISTENCE.md b/docs/SYNC_SKIPPED_ENTITY_PERSISTENCE.md new file mode 100644 index 0000000000..2467a290c4 --- /dev/null +++ b/docs/SYNC_SKIPPED_ENTITY_PERSISTENCE.md @@ -0,0 +1,243 @@ +# Sync: skipped entities and why they can be lost permanently + +Status: design note for a follow-up PR. Nothing here is implemented yet. + +This note exists because of a bug found while making sync read from a single +gateway (PE-9203). The bug **predates that change** and is invisible unless you +go looking for it, so it is written down here in full. + +--- + +## 1. The pre-existing silent drop + +When a sync cannot read an entity's metadata, that entity is dropped from the +user's local drive state, silently, and in the common case permanently. There is +no error surfaced, no retry across sessions, and no record that anything is +missing. + +### Evidence + +**A failed read becomes empty bytes, not an error.** +`ArweaveService._getEntityData` (`lib/services/arweave/arweave_service.dart`) +historically ended with: + +```dart +return getEntityDataFromNetwork(txId: txId).catchError((e) { + logger.e('Failed to get entity data from network', e); + return Uint8List(0); // <-- failure becomes empty bytes +}); +``` + +**The empty bytes then fail to parse, and the parse error is swallowed.** +In `createDriveEntityHistoryFromTransactions`, `FileEntity.fromTransaction(tx, +Uint8List(0), ...)` throws `EntityTransactionParseException`, which is caught by +the `on EntityTransactionParseException` handler and logged at warning level. +The entity is never added to `blockHistory`, so it never reaches the database. + +**The drive's watermark advances anyway.** +`_parseDriveTransactionsIntoDatabaseEntities` +(`lib/sync/domain/repositories/sync_repository.dart`) writes +`lastBlockHeight: Value(currentBlockHeight)` — both in the empty-transactions +branch and in `endOfBatchCallback` — with no knowledge that anything was +dropped. The next sync therefore starts *after* the block containing the item it +failed to read. + +**The look-back only helps within a session boundary.** +`_calculateSyncLastBlockHeight` rewinds by `kBlockHeightLookBack` +(`lib/sync/constants.dart`, currently 240 blocks ≈ 2 hours) — but only when +`_lastSync == null`: + +```dart +int _calculateSyncLastBlockHeight(int lastBlockHeight) { + if (_lastSync != null) { + return lastBlockHeight; // no rewind + } + return max(lastBlockHeight - kBlockHeightLookBack, 0); // rewind +} +``` + +`_lastSync` is a plain in-memory `DateTime?` on `_SyncRepository`. It is never +persisted. So the rewind happens once per app session and covers only the last +~2 hours of chain history. + +### What that means in practice + +| Situation | Recovered? | +|---|---| +| Item dropped, another sync runs in the same session | No — `_lastSync != null`, no rewind | +| Item dropped, app restarted within ~2h of that block | Yes — first sync of the session rewinds 240 blocks | +| Item dropped, app restarted more than ~2h later | **No — lost until a deep sync** | +| User triggers a deep sync | Yes — `syncDeep` passes `lastBlockHeight: 0` | + +Deep sync is the only reliable recovery, it is user-initiated, and nothing tells +the user they need it. A file uploaded from another device can therefore be +missing from this device indefinitely while the UI looks perfectly healthy. + +### What PE-9203 changed about this + +PE-9203 did not introduce this, but it does raise how often it fires. Sync reads +went from a 4-attempt waterfall (configured gateway + 2 GAR gateways + +arweave.net) to 2 attempts against the configured gateway only. Fewer attempts +means more skips. + +What PE-9203 *added* is the record: skipped transaction ids are now collected +and reported instead of being logged and forgotten. + +- `DriveEntityHistory.skippedTxIds` carries them out of `ArweaveService`. +- `_SyncRepository._skippedEntityTxIdsByDrive` accumulates them per drive. +- `SyncProgress.skippedEntityCount` / `.skippedEntityTxIdsByDrive` report them. +- `SyncCubit.lastSyncSkippedEntityTxIdsByDrive` retains them after the run. + +That record is **in-memory only**. It dies with the process, which is exactly +the gap this note proposes to close. + +--- + +## 2. Proposal: `sync_failed_entities` + +### Why a table, and not the watermark + +The tempting zero-schema fix is to clamp the watermark: on failure, don't +advance `drives.lastBlockHeight` past the oldest failed item's block height. +It uses an already-persisted field and guarantees a retry. + +It was rejected because it is unbounded. A genuinely dead transaction pins the +drive's watermark forever, so every subsequent sync re-queries from that height +— a compounding regression on the exact path PE-9203 exists to speed up. It also +cannot answer "which items failed" (only "something failed at or after block +N"), and it distorts snapshot range maths, since `HeightRange.difference` +derives the GQL sub-ranges from that same watermark. + +Distinguishing "transient" from "permanently unavailable" needs attempt history. +Attempt history needs persistence. Hence a table. + +### Schema + +```sql +CREATE TABLE sync_failed_entities ( + txId TEXT NOT NULL PRIMARY KEY, + driveId TEXT NOT NULL, + blockHeight INTEGER NOT NULL, + attempts INTEGER NOT NULL DEFAULT 1, + lastAttempt DATETIME NOT NULL, + lastError TEXT, + isTerminal BOOLEAN NOT NULL DEFAULT FALSE +) As SyncFailedEntity; + +CREATE INDEX idx_sync_failed_entities_drive ON sync_failed_entities(driveId); +``` + +Migration cost is small and mechanical: + +- add `lib/models/tables/sync_failed_entities.drift`, import it from `all.drift` + and from the `@DriftDatabase(include: {...})` set in `database.dart` +- bump `schemaVersion` 29 → 30 +- add `if (from < 30) { await m.createTable(syncFailedEntities); }`, matching the + existing style of the v19 and v21 migration blocks +- run `flutter pub run build_runner build --delete-conflicting-outputs` + +The pre-push hook (`lefthook/database_checker.sh`) only asserts that +`schemaVersion` increased when `lib/models/tables` changed. It does **not** +require a `drift_schemas/` JSON export — that directory is stale (last export +v19 against a live schemaVersion of 29), so no export is needed. + +### Retry: re-query the block, not the transaction + +To re-apply a recovered item we need its full GQL node (tags, block, owner) for +`createDriveEntityHistoryFromTransactions`. A bare transaction id is not enough, +and there is no existing query returning `...TransactionCommon` by ids — +`InfoOfTransactionsToBePinned` takes `ids:` but returns a narrower shape. + +Rather than add a GraphQL query and another generated-code surface, **store the +failed item's block height and union it back into the range that sync already +queries.** `_syncDrive` builds `totalRangeToQueryFor` as a multi-segment +`HeightRange`: + +```dart +final totalRangeToQueryFor = HeightRange( + rangeSegments: [ + Range(start: lastBlockHeight, end: currentBlockHeight), + ], +); +``` + +Adding `Range(start: h, end: h)` for each due failed height makes the normal +`DriveEntityHistory` query re-yield the failed transaction with all its tags, at +the cost of re-reading a few cheap siblings in the same block. No new query, no +new codegen beyond the table. + +Sketch: + +```dart +final dueHeights = await _driveDao.dueFailedEntityHeights(drive.id); +final totalRangeToQueryFor = HeightRange( + rangeSegments: [ + Range(start: lastBlockHeight, end: currentBlockHeight), + ...dueHeights.map((h) => Range(start: h, end: h)), + ], +); +``` + +`HeightRange.union` already normalises overlaps, so heights inside the primary +range cost nothing extra. + +### Backoff policy + +Retries are driven by the table, not by block height, so they are independent of +the 240-block look-back window. + +| Failure kind | Behaviour | +|---|---| +| `TransactionNotFound` (404) | Record with `isTerminal = true`. Never auto-retried — consistent with the fast-fail-on-404 stance in `SnapshotValidationService`. The row remains for the UI, and a deep sync clears it. | +| Transient (timeout, 5xx, network) | Retried at the start of the next sync for that drive, subject to backoff. | + +Backoff keyed on `attempts`, evaluated against `lastAttempt`: + +| attempts | earliest next retry | +|---|---| +| 1 | next sync | +| 2 | +5 minutes | +| 3 | +1 hour | +| 4 | +6 hours | +| 5 | stop; mark `isTerminal` | + +One integer, one timestamp, and a constant list of durations. No queue, no +isolate, no scheduler. The retry pass is: load the drive's due rows, union their +heights into the query range, let the existing path re-read them, then delete +the rows that succeeded and bump `attempts` on those that did not. + +### Lifecycle + +- **Write** on a failed metadata read, keyed by `txId` (upsert, incrementing + `attempts`). +- **Delete** as soon as that `txId` parses successfully in any later sync. +- **Clear per drive** on deep sync, which re-reads from block 0 anyway. +- **Never block a sync.** A failed retry is still just a skip. + +--- + +## 3. Follow-on work this unlocks + +1. **"Failed files" UI.** Design is pending. The data is already exposed + in-memory via `SyncProgress.skippedEntityTxIdsByDrive` and + `SyncCubit.lastSyncSkippedEntityTxIdsByDrive`; the table makes it durable and + makes a per-drive count queryable without a sync having just run. +2. **Automatic recovery**, removing the need for a user to know that deep sync + exists. +3. **Telemetry on skip rate**, which is the measurement needed before raising + `maxConcurrentDataFetches` (deliberately left at 5 in PE-9203 so that the + scheduling change and the request-rate change can be evaluated separately). + +## 4. Related code + +| Concern | Location | +|---|---| +| Failed read → `null` | `ArweaveService._getEntityData` | +| Parse error swallowed | `ArweaveService.createDriveEntityHistoryFromTransactions` | +| Watermark advance | `_SyncRepository._parseDriveTransactionsIntoDatabaseEntities` | +| Look-back rewind | `_SyncRepository._calculateSyncLastBlockHeight` | +| Look-back constant | `kBlockHeightLookBack`, `lib/sync/constants.dart` | +| Deep sync reset | `_SyncRepository.syncAllDrives`, `syncDeep ? 0 : ...` | +| Sync read path | `DataGatewayFallback.fetchDataForSync` | +| In-memory skip record | `SyncProgress`, `SyncCubit` | +| Range machinery | `lib/utils/snapshots/height_range.dart` | diff --git a/lib/main.dart b/lib/main.dart index 2e2f4f2a32..0b537ec941 100644 --- a/lib/main.dart +++ b/lib/main.dart @@ -487,7 +487,6 @@ class AppState extends State { batchProcessor: BatchProcessor(), snapshotValidationService: SnapshotValidationService( configService: configService, - arioSDK: ArioSDKFactory().create(), ), arnsRepository: _.read(), userPreferencesRepository: _.read(), diff --git a/lib/services/arweave/arweave_service.dart b/lib/services/arweave/arweave_service.dart index 95ffc1ea1e..d1ebb07a98 100644 --- a/lib/services/arweave/arweave_service.dart +++ b/lib/services/arweave/arweave_service.dart @@ -493,6 +493,43 @@ class ArweaveService { } } + /// Runs [task] for every index in `0..itemCount-1`, keeping at most + /// [concurrency] in flight and starting the next index as soon as any one + /// completes. + /// + /// This is a sliding window, not a chunked `Future.wait`. A chunked barrier + /// idles every other slot until the slowest member of the chunk returns, so + /// one slow or failing item costs the whole chunk its duration. + /// + /// [task] owns its error handling — a task that throws aborts the run. + @visibleForTesting + static Future runPooled({ + required int concurrency, + required int itemCount, + required Future Function(int index) task, + }) async { + if (itemCount <= 0) return; + + final workerCount = concurrency < 1 + ? 1 + : (concurrency < itemCount ? concurrency : itemCount); + + // Shared cursor. Claiming an index is synchronous, so two workers can + // never take the same one. + var nextIndex = 0; + + Future worker() async { + while (true) { + final i = nextIndex; + if (i >= itemCount) return; + nextIndex++; + await task(i); + } + } + + await Future.wait(List.generate(workerCount, (_) => worker())); + } + /// Get the metadata of transactions /// /// mounts the `blockHistory` @@ -507,43 +544,60 @@ class ArweaveService { int? currentBlockHeight, }) async { // Limit concurrent data fetches to avoid overwhelming the gateway. - // Uses chunked Future.wait — processes maxConcurrent at a time. + // + // Sliding window, not chunked Future.wait: workers pull the next index as + // soon as they finish, so exactly maxConcurrent fetches stay in flight. + // A chunked barrier would idle every other slot until the slowest member + // of the chunk returned — and with the 2-attempt sync retry, one dead + // transaction stalls its whole chunk for ~10s. final maxConcurrent = _configService.config.maxConcurrentDataFetches.clamp(1, 100); final entityDatas = List.filled(entityTxs.length, Uint8List(0)); - for (var start = 0; start < entityTxs.length; start += maxConcurrent) { - final end = (start + maxConcurrent < entityTxs.length) - ? start + maxConcurrent - : entityTxs.length; - - await Future.wait( - List.generate(end - start, (j) { - final i = start + j; - final entity = entityTxs[i].transactionCommonMixin; - final tags = HashMap.fromIterable( - entity.tags, - key: (tag) => tag.name, - value: (tag) => tag.value, - ); + /// Metadata reads that failed outright. These entities are skipped for + /// this sync — see the note on [_getEntityData] for why that can be a + /// permanent drop, and `docs/SYNC_SKIPPED_ENTITY_PERSISTENCE.md` for the + /// planned fix. Surfaced so callers can report them instead of losing them. + final skippedTxIds = []; + + await runPooled( + concurrency: maxConcurrent, + itemCount: entityTxs.length, + task: (i) async { + final entity = entityTxs[i].transactionCommonMixin; + final tags = HashMap.fromIterable( + entity.tags, + key: (tag) => tag.name, + value: (tag) => tag.value, + ); - if (driveKey != null && tags[EntityTag.cipherIv] == null) { - return Future.value(); - } - if (tags[EntityTag.entityType] == EntityTypeTag.snapshot) { - return Future.value(); - } + // Entities we never fetch. Leave entityDatas[i] at its empty default + // and release the slot immediately. + if (driveKey != null && tags[EntityTag.cipherIv] == null) { + return; + } + if (tags[EntityTag.entityType] == EntityTypeTag.snapshot) { + return; + } - return _getEntityData( - entityId: entity.id, - driveId: driveId, - isPrivate: driveKey != null, - ).then((data) { - entityDatas[i] = data; - }); - }), - ); - } + // _getEntityData never throws — a failed read skips only this entity + // and must never abort the run. + final data = await _getEntityData( + entityId: entity.id, + driveId: driveId, + isPrivate: driveKey != null, + ); + + if (data == null) { + skippedTxIds.add(entity.id); + return; + } + + // Positional write — callers rely on entityDatas aligning with + // entityTxs, so results are never appended. + entityDatas[i] = data; + }, + ); final metadataCache = await MetadataCache.fromCacheStore( await newSharedPreferencesCacheStore(), @@ -636,9 +690,17 @@ class ArweaveService { block.entities.removeWhere((e) => e!.ownerAddress != ownerAddress); } + if (skippedTxIds.isNotEmpty) { + logger.w( + 'Skipped ${skippedTxIds.length} entities in drive $driveId: their ' + 'metadata could not be read. They will not appear in this sync.', + ); + } + return DriveEntityHistory( blockHistory.isNotEmpty ? blockHistory.last.blockHeight : lastBlockHeight, blockHistory, + skippedTxIds: skippedTxIds, ); } @@ -657,7 +719,20 @@ class ArweaveService { return privateDriveTxs.isNotEmpty; } - Future _getEntityData({ + /// Returns the entity's metadata bytes, or `null` if they could not be read. + /// + /// KNOWN ISSUE — a `null` here is a silent, potentially permanent drop, and + /// it predates the single-gateway sync change. The caller substitutes empty + /// bytes, the entity fails to parse (swallowed at the `on + /// EntityTransactionParseException` in + /// [createDriveEntityHistoryFromTransactions]) and never reaches + /// `blockHistory` — while the drive's watermark advances regardless. Only a + /// user-initiated deep sync reliably recovers it. + /// + /// Full evidence and the planned fix (persist skipped items and retry them + /// across syncs) are in `docs/SYNC_SKIPPED_ENTITY_PERSISTENCE.md`. Until + /// then the tx ids are at least reported on `SyncProgress` rather than lost. + Future _getEntityData({ required String entityId, required String driveId, required bool isPrivate, @@ -674,9 +749,10 @@ class ArweaveService { return cachedData; } - return getEntityDataFromNetwork(txId: txId).catchError((e) { - logger.e('Failed to get entity data from network', e); - return Uint8List(0); + return getEntityDataFromNetwork(txId: txId).then((d) => d) + .catchError((e) { + logger.e('Failed to get entity data from network for tx $txId', e); + return null; }); } @@ -707,8 +783,13 @@ class ArweaveService { return null; } + /// Reads entity metadata for the **sync** path. + /// + /// Uses [DataGatewayFallback.fetchDataForSync] — configured gateway only, + /// one retry, one last-resort hop, no GAR and therefore no Solana RPC. + /// Download/preview/thumbnail/share paths keep the full waterfall. Future getEntityDataFromNetwork({required String txId}) async { - final Response data = await _gatewayFallback.fetchData(txId, client); + final Response data = await _gatewayFallback.fetchDataForSync(txId, client); return data.bodyBytes; } @@ -798,10 +879,34 @@ class ArweaveService { return firstTx; } + /// The drive signature, read through the multi-gateway waterfall. + /// + /// This is the login path (`ArDriveAuth`), where one unreachable gateway + /// must not cost someone their session. Sync reads the same signature + /// through [getDriveSignatureForDriveOnSync] instead. Future getDriveSignatureForDrive( Wallet wallet, String driveId, - ) async { + ) => + _getDriveSignature(wallet, driveId, forSync: false); + + /// The drive signature as **sync** reads it: the configured gateway only. + /// + /// Drive discovery needs this for every private drive whose key is not + /// already in memory. Routing it through the waterfall would have put the + /// fan-out back into the sync path by the side door, one drive at a time. + @visibleForTesting + Future getDriveSignatureForDriveOnSync( + Wallet wallet, + String driveId, + ) => + _getDriveSignature(wallet, driveId, forSync: true); + + Future _getDriveSignature( + Wallet wallet, + String driveId, { + required bool forSync, + }) async { // Drive signatures are immutable on-chain — cache permanently once fetched if (_cachedDriveSignatures.containsKey(driveId)) { return _cachedDriveSignatures[driveId]; @@ -810,7 +915,9 @@ class ArweaveService { final driveSignatureTx = await getDriveSignatureTxForDrive(wallet, driveId); final driveSignatureData = driveSignatureTx != null - ? await _gatewayFallback.fetchData(driveSignatureTx.id, client) + ? await (forSync + ? _gatewayFallback.fetchDataForSync(driveSignatureTx.id, client) + : _gatewayFallback.fetchData(driveSignatureTx.id, client)) : null; final driveSignature = @@ -831,11 +938,34 @@ class ArweaveService { final userAddress = await wallet.getAddress(); final driveTxs = await getUniqueUserDriveEntityTxs(userAddress); - final driveResponses = await Future.wait( - driveTxs.map((e) => _gatewayFallback - .fetchData(e.id, client) - .then((r) => r) - .catchError((_) => null)), + // Sync's drive-discovery phase, and its only caller is + // `_SyncRepository.updateUserDrives`. It reads the configured gateway + // only, like every other sync read: this fires once per drive + // transaction, so leaving it on the waterfall meant a user with a dozen + // drives opened a dozen fan-outs to GAR gateways on every sync - which + // is exactly the cost this change exists to remove. + // + // A drive whose metadata cannot be read is dropped from this pass, as + // before; the full sync below re-reads it. + // Bounded, not an unbounded `Future.wait` over every drive transaction. + // Now that this reads one gateway instead of fanning out across + // several, an unbounded burst is all aimed at that single host - and a + // user with many drives would open every connection at once. Same limit + // the metadata reads use. + final driveResponses = List.filled(driveTxs.length, null); + + await runPooled( + concurrency: + _configService.config.maxConcurrentDataFetches.clamp(1, 100), + itemCount: driveTxs.length, + task: (i) async { + try { + driveResponses[i] = + await _gatewayFallback.fetchDataForSync(driveTxs[i].id, client); + } catch (_) { + // A drive we cannot read is dropped from this pass, as before. + } + }, ); // Cache raw bytes for reuse by getLatestDriveEntityWithId (e.g., during @@ -869,7 +999,7 @@ class ArweaveService { final signatureType = DriveSignatureType.fromString(sigTypeTag); final driveSignature = signatureType == DriveSignatureType.v1 - ? await getDriveSignatureForDrive( + ? await getDriveSignatureForDriveOnSync( wallet, driveTx.getTag(EntityTag.driveId)!) : null; @@ -1420,15 +1550,20 @@ class ArweaveService { } final chunkStarts = [for (var i = 0; i < ids.length; i += chunkSize) i]; - // Process the chunks in bounded-concurrency batches. - for (var b = 0; b < chunkStarts.length; b += maxConcurrent) { - final batch = chunkStarts.skip(b).take(maxConcurrent); - try { - await Future.wait(batch.map(queryChunk)); - } catch (e) { - logger.e('Error getting transactions confirmations on exception', e); - rethrow; - } + + // A pool, not a chunked `Future.wait`. Batching these meant each batch + // waited for its slowest query before the next started, so one slow + // chunk left the other workers idle - the same barrier that cost the + // metadata reads above, on the confirmation queries this time. + try { + await runPooled( + concurrency: maxConcurrent, + itemCount: chunkStarts.length, + task: (i) => queryChunk(chunkStarts[i]), + ); + } catch (e) { + logger.e('Error getting transactions confirmations on exception', e); + rethrow; } } @@ -1780,7 +1915,16 @@ class DriveEntityHistory { /// A list of block entities, ordered by ascending block height. final List blockHistory; - DriveEntityHistory(this.lastBlockHeight, this.blockHistory); + /// Transactions whose metadata could not be read, and which were therefore + /// left out of [blockHistory]. Surfaced so the sync layer can count and + /// report them rather than dropping them silently. + final List skippedTxIds; + + DriveEntityHistory( + this.lastBlockHeight, + this.blockHistory, { + this.skippedTxIds = const [], + }); } /// The entities present in a particular block. diff --git a/lib/services/arweave/data_gateway_fallback.dart b/lib/services/arweave/data_gateway_fallback.dart index a51177598b..66ba1e54f1 100644 --- a/lib/services/arweave/data_gateway_fallback.dart +++ b/lib/services/arweave/data_gateway_fallback.dart @@ -11,13 +11,19 @@ import 'package:http/http.dart'; /// Provides data gateway fallback resilience. /// -/// Metadata fetches use serial waterfall (primary → GAR → arweave.net) to -/// avoid unnecessary requests during high-volume sync operations. +/// Two distinct read paths live here: /// -/// File downloads use hedged (staggered parallel) requests since they are -/// single user-initiated operations where latency matters. +/// * **Sync reads** ([fetchDataForSync]) use the configured gateway only, with +/// a single retry, then skip the item. No GAR, therefore no Solana RPC. +/// Sync is high volume, so per-item attempt count dominates. /// -/// Fallback order: primary → up to 2 GAR gateways → arweave.net +/// * **Everything else** — downloads, previews, thumbnails and shared links — +/// keeps the full waterfall (primary → up to 2 GAR gateways → arweave.net). +/// These are single user-initiated operations where a recipient with one +/// dead gateway must still get their file, so breadth beats latency. +/// +/// File downloads additionally use hedged (staggered parallel) requests since +/// they are single user-initiated operations where latency matters. class DataGatewayFallback { final ArioSDK _arioSDK; final Map _clientCache = {}; @@ -29,6 +35,17 @@ class DataGatewayFallback { static const _hedgeDelay = Duration(milliseconds: 1500); static const _downloadTimeout = Duration(seconds: 15); + /// Attempts made against the configured gateway by [fetchDataForSync]. + static const syncMaxAttempts = 2; + + /// Delay before the single same-gateway retry in [fetchDataForSync]. + static const _syncRetryDelay = Duration(milliseconds: 300); + + /// Upper bound for one sync read. By construction the attempts already sum + /// to ~10.3s; this only guards against an attempt that outlives its own + /// timeout. + static const _syncTotalFetchTimeout = Duration(seconds: 15); + /// Cached gateway list — shared with other services (e.g. /// SnapshotValidationService) to avoid duplicate Solana RPC calls. List? cachedGateways; @@ -51,6 +68,62 @@ class DataGatewayFallback { }); } + /// Fetch transaction data for **sync** metadata reads. + /// + /// Reads the configured gateway only: one attempt, one retry on a transient + /// failure, then give up so the caller can skip the item and carry on. Never + /// consults the GAR, so no Solana RPC is issued on the sync path. + /// + /// Sync issues hundreds of these per run, where per-item attempt count + /// dominates: the [fetchData] waterfall costs up to 4 serial attempts at 5s + /// each, so a slow or flaky primary turns into minutes of serial timeouts. + /// Worst case here is [syncMaxAttempts] attempts / ~10.3s. + /// + /// A 404 is not retried — the same host will not change its mind. + /// + /// Callers must treat a failure as "skip this item", and must record the tx + /// id so it is not lost. See the note on `ArweaveService._getEntityData` and + /// `docs/SYNC_SKIPPED_ENTITY_PERSISTENCE.md`. + /// + /// If every attempt 404s, throws [TransactionNotFound]. + Future fetchDataForSync(String txId, Arweave primaryClient) async { + return _syncFetch(txId, primaryClient) + .timeout(_syncTotalFetchTimeout, onTimeout: () { + logger.w('Total sync fetch timeout exceeded for tx $txId'); + throw Exception('Total sync fetch timeout exceeded for tx $txId'); + }); + } + + Future _syncFetch(String txId, Arweave primaryClient) async { + final gatewayName = primaryClient.api.gatewayUrl.host; + + for (var attempt = 1; attempt <= syncMaxAttempts; attempt++) { + try { + return await _tryGateway(primaryClient, txId); + } on _ErrorFromStatus catch (e) { + if (e.statusCode == 404) { + // Retrying the same host cannot turn a 404 into a 200. + logger.w('Gateway $gatewayName returned 404 for sync tx $txId'); + throw TransactionNotFound(txId); + } + logger.w('Gateway $gatewayName failed for sync tx $txId ' + '(attempt $attempt/$syncMaxAttempts): $e'); + } catch (e) { + logger.w('Gateway $gatewayName failed for sync tx $txId ' + '(attempt $attempt/$syncMaxAttempts): $e'); + } + + if (attempt < syncMaxAttempts) { + await Future.delayed(_syncRetryDelay); + } + } + + throw Exception( + 'Gateway $gatewayName failed for sync tx $txId ' + 'after $syncMaxAttempts attempts', + ); + } + Future _serialFetch(String txId, Arweave primaryClient) async { final clients = await _buildClientList(primaryClient); var all404 = true; diff --git a/lib/sync/data/snapshot_validation_service.dart b/lib/sync/data/snapshot_validation_service.dart index f404f0a837..3c68671532 100644 --- a/lib/sync/data/snapshot_validation_service.dart +++ b/lib/sync/data/snapshot_validation_service.dart @@ -1,30 +1,29 @@ import 'dart:async'; -import 'package:ardrive/services/arweave/data_gateway_fallback.dart'; import 'package:ardrive/services/config/config_service.dart'; import 'package:ardrive/utils/logger.dart'; import 'package:ardrive/utils/snapshots/snapshot_item.dart'; -import 'package:ario_sdk/ario_sdk.dart'; import 'package:http/http.dart' as http; class SnapshotValidationService { final ConfigService _configService; - final ArioSDK _arioSDK; static const _headTimeout = Duration(seconds: 5); - static const _garListTimeout = Duration(seconds: 3); static const _maxConcurrentValidations = 3; - /// Shared reference to [DataGatewayFallback] for reading/writing the - /// gateway cache. Set by SyncRepository before validation runs so both - /// services share one cache and avoid duplicate Solana RPC calls. - DataGatewayFallback? gatewayFallback; + /// Attempts against the configured gateway before a snapshot is rejected. + /// + /// Two, the same budget a sync metadata read gets - and the case for it is + /// stronger here. Losing a metadata read costs one entity; losing a snapshot + /// costs its entire block range, which sync then has to re-query over + /// GraphQL. The expensive failure should not be the one with fewer chances. + static const _maxAttempts = 2; + + static const _retryDelay = Duration(milliseconds: 300); SnapshotValidationService({ required ConfigService configService, - required ArioSDK arioSDK, - }) : _configService = configService, - _arioSDK = arioSDK; + }) : _configService = configService; Future> validateSnapshotItems( List snapshotItems, @@ -62,106 +61,57 @@ class SnapshotValidationService { return snapshotsVerified; } - /// Validates a snapshot is available on at least one gateway. + /// Validates a snapshot is available on the configured gateway. /// - /// 1. Try primary gateway with 1 retry on transient errors - /// 2. If primary fails, try 1 fallback gateway from the GAR list - /// 3. Accept if ANY gateway returns 200 + /// Single HEAD against the configured gateway; anything other than 200/302 + /// rejects the snapshot and sync falls back to GQL for that range, which is + /// correct (just slower) in every case. + /// + /// There is deliberately no GAR fallback here. It only ever ran for + /// transient errors, and it cost a Solana RPC via `ArioSDK.getGateways()` — + /// the sync path must not issue one. Rejecting a snapshot is cheap and safe; + /// paying a Solana round-trip to maybe save a GQL range is not. Future _validateSnapshot(String txId, String primaryUrl) async { - var primaryWas404 = false; - - // 1. Try primary gateway (1 attempt, no retry — fail fast) - try { - final response = await http - .head(Uri.parse('$primaryUrl/$txId')) - .timeout(_headTimeout); + for (var attempt = 1; attempt <= _maxAttempts; attempt++) { + try { + final response = await http + .head(Uri.parse('$primaryUrl/$txId')) + .timeout(_headTimeout); + + if (response.statusCode == 200 || response.statusCode == 302) { + return true; + } - if (response.statusCode == 200 || response.statusCode == 302) { - return true; - } + // A refusal the same host will repeat. Retrying spends time to be + // told the same thing. + if (_isNonRetryable(response.statusCode)) { + logger.w( + 'Snapshot $txId rejected: ' + 'non-retryable status ${response.statusCode}', + ); + return false; + } - if (_isNonRetryable(response.statusCode)) { logger.w( - 'Snapshot $txId rejected: ' - 'non-retryable status ${response.statusCode}', + 'Snapshot $txId: HEAD returned ${response.statusCode} ' + '(attempt $attempt/$_maxAttempts)', ); - return false; + } on TimeoutException { + logger.w('Snapshot $txId: HEAD timed out ' + '(attempt $attempt/$_maxAttempts)'); + } catch (e) { + logger.w('Snapshot $txId: HEAD error ' + '(attempt $attempt/$_maxAttempts): $e'); } - primaryWas404 = response.statusCode == 404; - - logger.d( - 'Snapshot $txId HEAD returned ${response.statusCode}', - ); - } on TimeoutException { - logger.d('Snapshot $txId HEAD timed out'); - } catch (e) { - logger.d('Snapshot $txId HEAD error: $e'); - } - - // 2. If primary returned 404, the snapshot likely doesn't exist. - // Skip the fallback — GAR list requires Solana RPC which may be - // unavailable (localhost, rate limits). Fail fast and fall back to GQL. - if (primaryWas404) { - logger.w('Snapshot $txId not found on primary (404), skipping fallback'); - return false; - } - - // 3. Primary had a transient error (timeout, 5xx) — try 1 fallback gateway. - // Read the shared gateway cache from DataGatewayFallback. If unavailable, - // fetch from Solana RPC once and cache the result (empty on failure). - try { - List gateways; - if (gatewayFallback != null && - gatewayFallback!.cachedGateways != null) { - gateways = gatewayFallback!.cachedGateways!; - } else { - try { - gateways = await _arioSDK - .getGateways() - .timeout(_garListTimeout, onTimeout: () => []); - // Store in shared cache if available - if (gatewayFallback != null) { - gatewayFallback!.cachedGateways = gateways; - } - } catch (e) { - // Solana RPC failed — cache empty list so we don't retry every call - logger.w('GAR gateway list unavailable, will not retry: $e'); - if (gatewayFallback != null) { - gatewayFallback!.cachedGateways = []; - } - gateways = []; - } - } - - if (gateways.isEmpty) return false; - - final primaryHost = Uri.parse(primaryUrl).host; - final fallback = gateways.firstWhere( - (gw) => gw.settings.fqdn != primaryHost, - orElse: () => gateways.first, - ); - - final response = await http - .head(Uri.parse('https://${fallback.settings.fqdn}/$txId')) - .timeout(_headTimeout); - - if (response.statusCode == 200 || response.statusCode == 302) { - logger.i( - 'Snapshot $txId validated via fallback ' - 'gateway ${fallback.settings.fqdn}', - ); - return true; + if (attempt < _maxAttempts) { + await Future.delayed(_retryDelay); } - - logger.w( - 'Snapshot $txId fallback gateway ${fallback.settings.fqdn} ' - 'returned ${response.statusCode}', - ); - } catch (e) { - logger.w('Snapshot $txId fallback validation failed: $e'); } + logger.w('Snapshot $txId rejected after $_maxAttempts attempts, ' + 'falling back to GQL for its range'); + return false; } diff --git a/lib/sync/domain/cubit/sync_cubit.dart b/lib/sync/domain/cubit/sync_cubit.dart index 92f3da7072..f11a0e287b 100644 --- a/lib/sync/domain/cubit/sync_cubit.dart +++ b/lib/sync/domain/cubit/sync_cubit.dart @@ -53,6 +53,27 @@ class SyncCubit extends Cubit { SyncProgress _syncProgress = SyncProgress.initial(); SyncCancellationToken? _currentSyncToken; + Map> _lastSyncSkippedEntityTxIdsByDrive = const {}; + + /// Transaction ids of entities the most recent sync could not read, keyed by + /// drive id. Retained on the cubit — not just on the terminal state — so it + /// survives a plain [SyncIdle] completion, which is the common case when + /// items are skipped but no drive outright fails. + /// + /// This is currently the only record that anything was dropped. A later pass + /// renders it as "failed files"; persisting and retrying these across syncs + /// is described in `docs/SYNC_SKIPPED_ENTITY_PERSISTENCE.md`. + Map> get lastSyncSkippedEntityTxIdsByDrive => + _lastSyncSkippedEntityTxIdsByDrive; + + int get lastSyncSkippedEntityCount => _lastSyncSkippedEntityTxIdsByDrive + .values + .fold(0, (sum, txIds) => sum + txIds.length); + + void _captureSkippedEntities(SyncProgress progress) { + _lastSyncSkippedEntityTxIdsByDrive = progress.skippedEntityTxIdsByDrive; + } + SyncCubit({ required ProfileCubit profileCubit, required ActivityCubit activityCubit, @@ -383,6 +404,8 @@ class SyncCubit extends Cubit { unawaited(_updateContext()); + _captureSkippedEntities(_syncProgress); + // Check if sync completed with errors (only for non-cancelled syncs) if (_syncProgress.hasErrors) { logger.w('Sync completed with ${_syncProgress.failedQueries} errors'); @@ -391,6 +414,8 @@ class SyncCubit extends Cubit { totalDrives: _syncProgress.drivesCount, failedDriveIds: _syncProgress.failedDriveIds, errorMessages: _syncProgress.errorMessages, + skippedEntityCount: _syncProgress.skippedEntityCount, + skippedEntityTxIdsByDrive: _syncProgress.skippedEntityTxIdsByDrive, )); } else { emit(SyncIdle()); @@ -526,6 +551,8 @@ class SyncCubit extends Cubit { _promptToSnapshotBloc.add(const SyncRunning(isRunning: false)); + _captureSkippedEntities(_syncProgress); + // Check if sync completed with errors if (_syncProgress.hasErrors) { logger.w('Single drive sync completed with errors'); @@ -534,6 +561,8 @@ class SyncCubit extends Cubit { totalDrives: _syncProgress.drivesCount, failedDriveIds: _syncProgress.failedDriveIds, errorMessages: _syncProgress.errorMessages, + skippedEntityCount: _syncProgress.skippedEntityCount, + skippedEntityTxIdsByDrive: _syncProgress.skippedEntityTxIdsByDrive, )); } else { emit(SyncIdle()); diff --git a/lib/sync/domain/cubit/sync_state.dart b/lib/sync/domain/cubit/sync_state.dart index 2dd687f9d8..0ac5a28b3e 100644 --- a/lib/sync/domain/cubit/sync_state.dart +++ b/lib/sync/domain/cubit/sync_state.dart @@ -47,13 +47,27 @@ class SyncCompleteWithErrors extends SyncState { final List failedDriveIds; final Map errorMessages; + /// Entities dropped from this sync because their metadata could not be read. + /// See [SyncCubit.lastSyncSkippedEntityTxIdsByDrive]. + final int skippedEntityCount; + final Map> skippedEntityTxIdsByDrive; + SyncCompleteWithErrors({ required this.failedDrives, required this.totalDrives, required this.failedDriveIds, required this.errorMessages, + this.skippedEntityCount = 0, + this.skippedEntityTxIdsByDrive = const {}, }); @override - List get props => [failedDrives, totalDrives, failedDriveIds, errorMessages]; + List get props => [ + failedDrives, + totalDrives, + failedDriveIds, + errorMessages, + skippedEntityCount, + skippedEntityTxIdsByDrive, + ]; } diff --git a/lib/sync/domain/repositories/sync_repository.dart b/lib/sync/domain/repositories/sync_repository.dart index 92d980c094..2f4e0a9e54 100644 --- a/lib/sync/domain/repositories/sync_repository.dart +++ b/lib/sync/domain/repositories/sync_repository.dart @@ -152,6 +152,44 @@ class _SyncRepository implements SyncRepository { final Map _ghostFolders = {}; final Set _folderIds = {}; + /// Entities skipped this sync because their metadata could not be read, + /// keyed by drive id. Reported on [SyncProgress] so a later pass can surface + /// "failed files" in the UI. + /// + /// In-memory only — cleared at the start of each sync. Persisting these (so + /// they are retried across syncs instead of relying on the 240-block + /// look-back) is the follow-up described in + /// `docs/SYNC_SKIPPED_ENTITY_PERSISTENCE.md`. + final Map> _skippedEntityTxIdsByDrive = {}; + + int get _skippedEntityCount => _skippedEntityTxIdsByDrive.values + .fold(0, (sum, txIds) => sum + txIds.length); + + Map> get _skippedEntityTxIdsByDriveSnapshot => { + for (final entry in _skippedEntityTxIdsByDrive.entries) + entry.key: [...entry.value], + }; + + void _recordSkippedEntities(String driveId, List txIds) { + if (txIds.isEmpty) return; + _skippedEntityTxIdsByDrive + .putIfAbsent(driveId, () => {}) + .addAll(txIds); + } + + void _logSkippedEntities() { + if (_skippedEntityTxIdsByDrive.isEmpty) return; + logger.w( + 'Sync skipped $_skippedEntityCount entities across ' + '${_skippedEntityTxIdsByDrive.length} drive(s); their metadata could ' + 'not be read from the configured gateway. ' + 'See docs/SYNC_SKIPPED_ENTITY_PERSISTENCE.md', + ); + for (final entry in _skippedEntityTxIdsByDrive.entries) { + logger.w('Drive ${entry.key} skipped tx ids: ${entry.value.join(', ')}'); + } + } + /// Maximum number of transactions to hold in memory during streaming sync. /// Larger values = better throughput, higher memory usage /// Smaller values = lower memory usage, more frequent DB commits @@ -196,6 +234,7 @@ class _SyncRepository implements SyncRepository { // Clear shared state from any previous sync to prevent stale data _ghostFolders.clear(); _folderIds.clear(); + _skippedEntityTxIdsByDrive.clear(); // The address of the currently logged-in wallet. All pending transactions // are uploads made by this wallet, so scoping the status query by it lets @@ -242,10 +281,6 @@ class _SyncRepository implements SyncRepository { ), ); - // Share gateway fallback reference so snapshot validation and data fetching - // use the same gateway cache (avoids duplicate Solana RPC calls) - _snapshotValidationService.gatewayFallback = _arweave.gatewayFallback; - // Probe for drive activity to skip unchanged drives. // Partition drives: never-synced drives always need full sync and would // poison the probe's minBlockHeight to 0 (causing it to query from genesis, @@ -639,8 +674,11 @@ class _SyncRepository implements SyncRepository { syncProgress = syncProgress.copyWith( progress: 1.0, statusMessage: 'Sync complete', + skippedEntityCount: _skippedEntityCount, + skippedEntityTxIdsByDrive: _skippedEntityTxIdsByDriveSnapshot, ); syncProgressController.add(syncProgress); + _logSkippedEntities(); // Close the controller when everything is done logger.d('Sync completed successfully, closing controller'); @@ -740,6 +778,7 @@ class _SyncRepository implements SyncRepository { // Clear shared state from any previous sync to prevent stale data _ghostFolders.clear(); _folderIds.clear(); + _skippedEntityTxIdsByDrive.clear(); // Get the specific drive final drive = @@ -923,8 +962,11 @@ class _SyncRepository implements SyncRepository { syncProgress = syncProgress.copyWith( progress: 1.0, statusMessage: 'Sync complete', + skippedEntityCount: _skippedEntityCount, + skippedEntityTxIdsByDrive: _skippedEntityTxIdsByDriveSnapshot, ); syncProgressController.add(syncProgress); + _logSkippedEntities(); logger.d('Single drive sync completed successfully, closing controller'); await syncProgressController.close(); @@ -1919,6 +1961,8 @@ class _SyncRepository implements SyncRepository { currentBlockHeight: currentBlockHeight, ); + _recordSkippedEntities(drive.id, entityHistory.skippedTxIds); + // Create entries for all the new revisions of file and folders in this drive. final newEntities = entityHistory.blockHistory .map((b) => b.entities) diff --git a/lib/sync/domain/sync_progress.dart b/lib/sync/domain/sync_progress.dart index 92958c80f4..b43f2fca6d 100644 --- a/lib/sync/domain/sync_progress.dart +++ b/lib/sync/domain/sync_progress.dart @@ -19,6 +19,8 @@ class SyncProgress extends LinearProgress { this.statusMessage, this.isSingleDriveSync = false, this.driveName, + this.skippedEntityCount = 0, + this.skippedEntityTxIdsByDrive = const {}, }); factory SyncProgress.initial() { @@ -73,8 +75,23 @@ class SyncProgress extends LinearProgress { final bool isSingleDriveSync; // true if syncing a single drive final String? driveName; // name of the drive being synced (for single drive sync) + /// Number of entities left out of this sync because their metadata could not + /// be read from the configured gateway. + final int skippedEntityCount; + + /// The skipped entities' transaction ids, keyed by drive id. This is the + /// only record that anything was dropped — a later pass surfaces these as + /// "failed files" in the UI. See `docs/SYNC_SKIPPED_ENTITY_PERSISTENCE.md`. + final Map> skippedEntityTxIdsByDrive; + + /// Flat list of every skipped transaction id, drive association discarded. + List get skippedEntityTxIds => + [...skippedEntityTxIdsByDrive.values.expand((txIds) => txIds)]; + // Helper getters bool get hasErrors => failedQueries > 0; + + bool get hasSkippedEntities => skippedEntityCount > 0; bool get isPartialSync => hasErrors && progress >= 1.0; bool get isCompleteWithErrors => progress >= 1.0 && hasErrors; @@ -91,6 +108,8 @@ class SyncProgress extends LinearProgress { Object? statusMessage = _absent, bool? isSingleDriveSync, Object? driveName = _absent, + int? skippedEntityCount, + Map>? skippedEntityTxIdsByDrive, }) { return SyncProgress( numberOfEntities: numberOfEntities ?? this.numberOfEntities, @@ -109,6 +128,9 @@ class SyncProgress extends LinearProgress { isSingleDriveSync: isSingleDriveSync ?? this.isSingleDriveSync, driveName: driveName == _absent ? this.driveName : driveName as String?, + skippedEntityCount: skippedEntityCount ?? this.skippedEntityCount, + skippedEntityTxIdsByDrive: + skippedEntityTxIdsByDrive ?? this.skippedEntityTxIdsByDrive, ); } } diff --git a/test/services/arweave/data_gateway_fallback_test.dart b/test/services/arweave/data_gateway_fallback_test.dart new file mode 100644 index 0000000000..b9765cad3d --- /dev/null +++ b/test/services/arweave/data_gateway_fallback_test.dart @@ -0,0 +1,270 @@ +import 'package:ardrive/services/arweave/arweave_service.dart'; +import 'package:ardrive/services/arweave/data_gateway_fallback.dart'; +import 'package:ario_sdk/ario_sdk.dart'; +import 'package:arweave/arweave.dart'; +import 'package:flutter_test/flutter_test.dart'; +import 'package:http/http.dart'; +import 'package:mocktail/mocktail.dart'; + +class _MockArioSDK extends Mock implements ArioSDK {} + +class _MockArweave extends Mock implements Arweave {} + +class _MockArweaveApi extends Mock implements ArweaveApi {} + +void main() { + late _MockArioSDK arioSDK; + late _MockArweave primaryClient; + late _MockArweaveApi primaryApi; + late DataGatewayFallback fallback; + + const txId = 'a-transaction-id'; + + setUp(() { + arioSDK = _MockArioSDK(); + primaryClient = _MockArweave(); + primaryApi = _MockArweaveApi(); + + when(() => primaryClient.api).thenReturn(primaryApi); + when(() => primaryApi.gatewayUrl) + .thenReturn(Uri.parse('https://configured-gateway.example')); + // Returning an empty list keeps the waterfall from making real network + // calls while still letting us verify whether the GAR was consulted. + when(() => arioSDK.getGateways()).thenAnswer((_) async => []); + + fallback = DataGatewayFallback(arioSDK: arioSDK); + }); + + group('fetchDataForSync — single configured gateway', () { + test('returns the response without retrying when the first attempt ' + 'succeeds', () async { + when(() => primaryApi.getSandboxedTx(txId)) + .thenAnswer((_) async => Response('metadata', 200)); + + final response = await fallback.fetchDataForSync(txId, primaryClient); + + expect(response.statusCode, 200); + expect(response.body, 'metadata'); + verify(() => primaryApi.getSandboxedTx(txId)).called(1); + }); + + test('retries a transient failure exactly once, then succeeds', () async { + var attempts = 0; + when(() => primaryApi.getSandboxedTx(txId)).thenAnswer((_) async { + attempts++; + if (attempts == 1) throw Exception('connection reset'); + return Response('metadata', 200); + }); + + final response = await fallback.fetchDataForSync(txId, primaryClient); + + expect(response.statusCode, 200); + expect(attempts, 2); + }); + + test('gives up after two attempts on a persistent transient failure, so ' + 'the caller can skip the item', () async { + when(() => primaryApi.getSandboxedTx(txId)) + .thenAnswer((_) async => Response('boom', 500)); + + await expectLater( + fallback.fetchDataForSync(txId, primaryClient), + throwsA(isA()), + ); + + verify(() => primaryApi.getSandboxedTx(txId)) + .called(DataGatewayFallback.syncMaxAttempts); + expect(DataGatewayFallback.syncMaxAttempts, 2); + }); + + test('does not retry a 404 — the same host will not change its mind', + () async { + when(() => primaryApi.getSandboxedTx(txId)) + .thenAnswer((_) async => Response('not found', 404)); + + await expectLater( + fallback.fetchDataForSync(txId, primaryClient), + throwsA(isA()), + ); + + verify(() => primaryApi.getSandboxedTx(txId)).called(1); + }); + + test('never consults the GAR, so no Solana RPC is issued on the sync path', + () async { + when(() => primaryApi.getSandboxedTx(txId)) + .thenAnswer((_) async => Response('boom', 500)); + + await expectLater( + fallback.fetchDataForSync(txId, primaryClient), + throwsA(isA()), + ); + + verifyNever(() => arioSDK.getGateways()); + expect(fallback.cachedGateways, isNull); + }); + + test('never falls back to another host', () async { + when(() => primaryApi.getSandboxedTx(txId)) + .thenAnswer((_) async => Response('boom', 503)); + + await expectLater( + fallback.fetchDataForSync(txId, primaryClient), + throwsA(isA()), + ); + + // Only the configured gateway's api was ever asked for a transaction. + verify(() => primaryApi.getSandboxedTx(txId)).called(2); + verifyNever(() => arioSDK.getGateways()); + }); + }); + + group('waterfall preserved for non-sync paths', () { + test('fetchData still consults the GAR (download/preview resilience)', + () async { + when(() => primaryApi.getSandboxedTx(txId)) + .thenAnswer((_) async => Response('metadata', 200)); + + await fallback.fetchData(txId, primaryClient); + + verify(() => arioSDK.getGateways()).called(1); + }); + + test('downloadWithFallback still consults the GAR', () async { + // The download itself will fail against the fake client; all we assert + // is that the client list was built from the GAR. + await expectLater( + fallback.downloadWithFallback( + txId: txId, + primaryClient: primaryClient, + ), + throwsA(isA()), + ); + + verify(() => arioSDK.getGateways()).called(1); + }); + + test('fetchData and fetchDataForSync differ in GAR usage for the same ' + 'transaction', () async { + when(() => primaryApi.getSandboxedTx(txId)) + .thenAnswer((_) async => Response('metadata', 200)); + + await fallback.fetchDataForSync(txId, primaryClient); + verifyNever(() => arioSDK.getGateways()); + + await fallback.fetchData(txId, primaryClient); + verify(() => arioSDK.getGateways()).called(1); + }); + + test('a drive signature read on the sync path never reaches the GAR', + () async { + // Drive discovery fetches this for every private drive whose key is not + // already in memory. It used to go through `getDriveSignatureForDrive`, + // which is the login path and uses the waterfall — putting the fan-out + // back into sync one drive at a time. + const signatureTxId = 'gPzMbUCLZ_1lJ6mCLQK4vGLtiOTKn1TfnCU8gLuLBLM'; + + when(() => primaryApi.getSandboxedTx(signatureTxId)) + .thenAnswer((_) async => Response('signature', 200)); + + await fallback.fetchDataForSync(signatureTxId, primaryClient); + + verifyNever(() => arioSDK.getGateways()); + expect(fallback.cachedGateways, isNull); + }); + }); + + group('ArweaveService.runPooled', () { + test('writes results positionally when tasks complete out of order', + () async { + // Later indices finish first, so appending would scramble the output. + final results = List.filled(6, null); + final delaysMs = [60, 50, 40, 30, 20, 10]; + + await ArweaveService.runPooled( + concurrency: 3, + itemCount: 6, + task: (i) async { + await Future.delayed(Duration(milliseconds: delaysMs[i])); + results[i] = i; + }, + ); + + expect(results, [0, 1, 2, 3, 4, 5]); + }); + + test('keeps at most `concurrency` tasks in flight', () async { + var inFlight = 0; + var maxObserved = 0; + + await ArweaveService.runPooled( + concurrency: 4, + itemCount: 20, + task: (i) async { + inFlight++; + maxObserved = inFlight > maxObserved ? inFlight : maxObserved; + await Future.delayed(const Duration(milliseconds: 5)); + inFlight--; + }, + ); + + expect(maxObserved, 4); + expect(inFlight, 0); + }); + + test('runs every index exactly once', () async { + final seen = []; + + await ArweaveService.runPooled( + concurrency: 5, + itemCount: 50, + task: (i) async { + await Future.delayed(const Duration(milliseconds: 1)); + seen.add(i); + }, + ); + + expect(seen.length, 50); + expect(seen.toSet().length, 50); + }); + + test('one slow task does not stall unrelated items', () async { + // Index 0 is slow; with a chunked barrier the other workers would idle + // until it finished. Pooled, they should stream through it. + final completionOrder = []; + + await ArweaveService.runPooled( + concurrency: 2, + itemCount: 6, + task: (i) async { + await Future.delayed( + Duration(milliseconds: i == 0 ? 120 : 5), + ); + completionOrder.add(i); + }, + ); + + // The slow item finishes last despite being claimed first. + expect(completionOrder.last, 0); + expect(completionOrder.length, 6); + }); + + test('clamps worker count to itemCount and handles an empty workload', + () async { + var ran = 0; + + await ArweaveService.runPooled( + concurrency: 10, + itemCount: 2, + task: (_) async => ran++, + ); + expect(ran, 2); + + await ArweaveService.runPooled( + concurrency: 10, + itemCount: 0, + task: (_) async => fail('should not run'), + ); + }); + }); +} diff --git a/test/sync/data/snapshot_validation_service_test.dart b/test/sync/data/snapshot_validation_service_test.dart new file mode 100644 index 0000000000..25f41bafe1 --- /dev/null +++ b/test/sync/data/snapshot_validation_service_test.dart @@ -0,0 +1,72 @@ +import 'package:ardrive/services/config/app_config.dart'; +import 'package:ardrive/services/config/config_service.dart'; +import 'package:ardrive/services/config/selected_gateway.dart'; +import 'package:ardrive/sync/data/snapshot_validation_service.dart'; +import 'package:ardrive/utils/snapshots/snapshot_item.dart'; +import 'package:ario_sdk/ario_sdk.dart'; +import 'package:flutter_test/flutter_test.dart'; +import 'package:mocktail/mocktail.dart'; + +class _MockConfigService extends Mock implements ConfigService {} + +class _MockArioSDK extends Mock implements ArioSDK {} + +/// The only member [SnapshotValidationService] reads off a snapshot item. +/// +/// A real [SnapshotItemOnChain] would drag a GraphQL node and a byte source in +/// with it, none of which the validation path touches. +class _FakeSnapshotItem extends Fake implements SnapshotItem { + @override + final String txId = 'Iw3hSMB1kQ9Vpp5rUwx5Z4kv7cKzYwzZ_-7QK4mUuTc'; +} + +void main() { + group('SnapshotValidationService', () { + late _MockConfigService configService; + + setUp(() { + configService = _MockConfigService(); + when(() => configService.config).thenReturn( + AppConfig( + // Unroutable host so the HEAD fails fast without touching a real + // gateway; validation must still complete and simply reject. + arweaveGatewayForDataRequest: const SelectedGateway( + label: 'test', + url: 'https://localhost:1', + ), + allowedDataItemSizeForTurbo: 1, + stripePublishableKey: '', + ), + ); + }); + + test('constructor takes no ArioSDK — the GAR fallback is gone', () { + // A compile-time guarantee: this only builds because `arioSDK` is no + // longer a required parameter. + final service = SnapshotValidationService(configService: configService); + expect(service, isA()); + }); + + test('rejects a snapshot without ever asking for a gateway list', + () async { + final arioSDK = _MockArioSDK(); + final service = SnapshotValidationService(configService: configService); + + // A nonempty list, so validation actually runs: an empty one returns + // before the loop and would assert nothing at all. + final verified = await service.validateSnapshotItems([ + _FakeSnapshotItem(), + ]); + + // The configured gateway is unreachable, so the snapshot is rejected - + // and on `dev` this is the point where the service would have reached + // for the GAR list to try a second gateway. + expect(verified, isEmpty); + + // Vacuous on its own, since the service is never handed this SDK. It is + // the compile-time signature above that proves the branch is gone; this + // documents the intent at the call site. + verifyZeroInteractions(arioSDK); + }); + }); +} diff --git a/test/sync/domain/sync_progress_skipped_entities_test.dart b/test/sync/domain/sync_progress_skipped_entities_test.dart new file mode 100644 index 0000000000..48122756c3 --- /dev/null +++ b/test/sync/domain/sync_progress_skipped_entities_test.dart @@ -0,0 +1,75 @@ +import 'package:ardrive/services/arweave/arweave_service.dart'; +import 'package:ardrive/sync/domain/sync_progress.dart'; +import 'package:flutter_test/flutter_test.dart'; + +void main() { + group('DriveEntityHistory.skippedTxIds', () { + test('defaults to empty so existing callers are unaffected', () { + final history = DriveEntityHistory(100, []); + expect(history.skippedTxIds, isEmpty); + }); + + test('carries the skipped transaction ids out of the arweave service', () { + final history = DriveEntityHistory(100, [], skippedTxIds: ['tx-a']); + expect(history.skippedTxIds, ['tx-a']); + }); + }); + + group('SyncProgress skipped-entity reporting', () { + test('defaults to nothing skipped', () { + final progress = SyncProgress.initial(); + + expect(progress.skippedEntityCount, 0); + expect(progress.skippedEntityTxIdsByDrive, isEmpty); + expect(progress.skippedEntityTxIds, isEmpty); + expect(progress.hasSkippedEntities, isFalse); + }); + + test('reports the count and the tx ids keyed by drive', () { + final progress = SyncProgress.initial().copyWith( + skippedEntityCount: 3, + skippedEntityTxIdsByDrive: { + 'drive-1': ['tx-a', 'tx-b'], + 'drive-2': ['tx-c'], + }, + ); + + expect(progress.skippedEntityCount, 3); + expect(progress.hasSkippedEntities, isTrue); + expect(progress.skippedEntityTxIdsByDrive['drive-1'], ['tx-a', 'tx-b']); + expect(progress.skippedEntityTxIdsByDrive['drive-2'], ['tx-c']); + expect( + progress.skippedEntityTxIds, + containsAll(['tx-a', 'tx-b', 'tx-c']), + ); + }); + + test('skipped entities are independent of drive-level query failures', () { + // A drive can sync "successfully" while still dropping entities, so the + // skip count must not be inferred from failedQueries. + final progress = SyncProgress.initial().copyWith( + skippedEntityCount: 1, + skippedEntityTxIdsByDrive: { + 'drive-1': ['tx-a'], + }, + ); + + expect(progress.hasErrors, isFalse); + expect(progress.hasSkippedEntities, isTrue); + }); + + test('copyWith preserves skipped data when not overridden', () { + final progress = SyncProgress.initial().copyWith( + skippedEntityCount: 2, + skippedEntityTxIdsByDrive: { + 'drive-1': ['tx-a', 'tx-b'], + }, + ); + + final later = progress.copyWith(progress: 1.0); + + expect(later.skippedEntityCount, 2); + expect(later.skippedEntityTxIdsByDrive['drive-1'], ['tx-a', 'tx-b']); + }); + }); +}