sync_queue

A durable offline-first mutation queue and sync engine core for Flutter apps. sync_queue stores serializable operations, coordinates retries and per-entity ordering, surfaces conflicts, and exposes UI-friendly sync state.

It is not a generic callback queue for queue.add(() async {}). It is meant for local-first apps that need queued API mutations to survive app restarts, network failures, and conflict resolution flows.

Features

  • Queue create, update, delete, or custom operations.
  • Use create, update, and delete enqueue helpers for common mutations.
  • Reject duplicate operation ids before they overwrite queued work.
  • Keep only the latest pending mutation for noisy edit flows.
  • Update pending operations before they are sent.
  • Pair optimistic local changes with queue commit rollback.
  • Persist queue records behind a storage interface.
  • Send operations through an app-owned transport adapter.
  • Bound transport sends with an optional timeout.
  • Retry failed operations with exponential backoff.
  • Add jitter to retry delays to avoid retry bursts.
  • Honor transport-provided retry delays.
  • Automatically retry due operations on schedule.
  • Schedule retry timers only for runnable queued work.
  • Read the next runnable retry time for custom schedulers.
  • Inspect drain summaries after each sync pass.
  • Limit drain passes for batched background sync.
  • Preserve operation order for each entity during drain passes.
  • Watch engine lifecycle state for sync spinners and debug panels.
  • Coalesce full drain requests that arrive while another drain is active.
  • Drain one due operation without draining unrelated work.
  • Drain due work for a specific entity.
  • Recover interrupted syncing operations after app restarts.
  • Pause automatic draining while offline.
  • Drain pending operations when connectivity returns.
  • Serialize operations and records for durable stores.
  • Adapt any JSON-capable persistence layer through JsonSyncStore.
  • Let JSON storage adapters provide optimized pending queries.
  • Surface conflicts instead of hiding them.
  • Retry failed operations after user action.
  • Retry all failed operations or only failed work for one entity.
  • Discard all failed operations or only failed work for one entity.
  • Discard queued operations before they are sent.
  • Discard pending work for a specific entity.
  • Discard queued work with an application-owned predicate.
  • Serialize queue mutations with drain state transitions for race-free cleanup.
  • Watch per-entity sync status from Flutter UI.
  • Inspect queued records for a specific entity.
  • Filter queued records by entity and lifecycle status.
  • Read aggregated entity sync state for badges and status rows.
  • Read queue-wide snapshots for global indicators and debug panels.
  • Watch combined sync state for app bars, badges, and debug panels.
  • Build Flutter UI with SyncStateBuilder and SyncEntityStateBuilder.
  • Show prebuilt sync badges with SyncQueueStatusBadge and SyncEntityStatusBadge.
  • Inspect queued work in Flutter with SyncQueueDebugView.

Getting Started

sync_queue intentionally keeps storage and networking abstract. Your app owns the API client, local database, and connectivity source; the package owns queue state, retry timing, conflict surfacing, and UI-friendly snapshots.

Use InMemorySyncStore for tests and prototypes. For production persistence, use the official sync_queue_drift adapter, or bring your own store through JsonSyncStore or a custom SyncStore.

See example/main.dart for a complete fake API flow with offline queuing, optimistic updates, failed-operation retry, and conflict resolution.

Guides

  • Architecture explains the boundary between storage, transport, the engine, and application-owned business rules.
  • Conflict resolution shows how transports report conflicts and how applications resolve them with retry, discard, or fail decisions.
  • Queue management roadmap outlines future infrastructure-level compaction and cleanup APIs without adding domain logic.
  • Concurrency and storage documents the single-owner engine model and the expectations for multi-writer stores.

Quick Start

final store = InMemorySyncStore();
final connectivity = ManualSyncConnectivity();
final engine = SyncEngine(
  store: store,
  transport: MyApiSyncTransport(),
  connectivity: connectivity,
  retryPolicy: const RetryPolicy(jitterFactor: 0.2),
  sendTimeout: const Duration(seconds: 30),
);

await engine.enqueueUpdate(
  entity: const SyncEntityRef(type: 'task', id: 'task-2'),
  payload: const {'title': 'Ship the package'},
);

Transport Adapter

Implement SyncTransport with your API client. Return success when the server accepts the operation, failure when it should retry or stop, and conflict when your app needs a merge decision.

class TaskSyncTransport implements SyncTransport {
  TaskSyncTransport(this.api);

  final TaskApi api;

  @override
  Future<SyncResult> send(SyncOperation operation) async {
    try {
      await api.sendTaskMutation(
        id: operation.entity.id,
        type: operation.type.wireName,
        payload: operation.payload,
      );
      return const SyncResult.success();
    } on VersionConflict catch (error) {
      return SyncResult.conflict(
        message: 'Server version changed.',
        local: operation.payload,
        remote: error.remotePayload,
      );
    } on RateLimited catch (error) {
      return SyncResult.failure(
        SyncFailure(
          message: 'Rate limited.',
          code: 'rate_limited',
          retryAfter: error.retryAfter,
        ),
      );
    }
  }
}

Storage Adapter

For durable production persistence, use sync_queue_drift, the official Drift/SQLite store. Queued mutations survive app restarts without writing a persistence layer yourself:

final store = SyncQueueDriftStore(NativeDatabase.createInBackground(file));
final engine = SyncEngine(
  store: store,
  transport: TaskSyncTransport(api),
);

Use JsonSyncStore when your app already has a persistence layer that can store maps, JSON blobs, or encoded records. Implement SyncJsonQueryStorage if your database can efficiently query due pending records.

final engine = SyncEngine(
  store: JsonSyncStore(MyJsonStorage(database)),
  transport: TaskSyncTransport(api),
);

For tests, InMemorySyncStore is usually enough:

final engine = SyncEngine(
  store: InMemorySyncStore(),
  transport: FakeSyncTransport(),
);

Queue Mutations

Operation ids must be unique while they are in the queue. Duplicate ids are rejected before they can replace existing work.

await engine.enqueueCreate(
  entity: const SyncEntityRef(type: 'task', id: 'task-1'),
  payload: const {'title': 'New task'},
);

await engine.enqueueLatestMutation(
  entity: const SyncEntityRef(type: 'task', id: 'task-1'),
  type: SyncOperationType.update,
  payload: const {'title': 'Only the latest pending update stays queued'},
);

await SyncOptimistic.run(
  apply: () => updateLocalTaskTitle('task-1', 'Optimistic title'),
  commit: () => engine.enqueueUpdate(
    entity: const SyncEntityRef(type: 'task', id: 'task-1'),
    payload: const {'title': 'Optimistic title'},
  ),
  rollback: (_, _) => updateLocalTaskTitle('task-1', 'Previous title'),
);

Draining

Operations for the same entity are drained in creation order. If an older operation is failed, conflicted, syncing, or waiting for a future retry, newer operations for that entity wait behind it while unrelated entities can continue.

final drain = await engine.drain();
final batch = await engine.drain(maxOperations: 25);

if (batch.shouldContinue) {
  scheduleAnotherSyncPass();
}

await engine.drainEntity(
  const SyncEntityRef(type: 'task', id: 'task-1'),
);

await engine.drainOperation('operation-1');

await engine.recoverInterruptedOperations(
  staleAfter: const Duration(minutes: 5),
);

final nextRetryAt = await engine.readNextRetryAt();

Queue Semantics

sync_queue is a durable sync operation queue, not a generic priority job scheduler. Its ordering and failure rules are tuned for local-first API mutations:

  • Transport exceptions are caught and stored on the operation as SyncFailure. A drain returns SyncDrainResult with failure counts instead of throwing transport errors to the caller.
  • If an operation fails, newer operations for the same entity wait behind it. Operations for unrelated entities can continue in the same drain pass.
  • If a job enqueues more work with the default syncImmediately: true while a full drain is active, the drain request is coalesced and the full drain reruns after the current pass. With syncImmediately: false, the nested operation waits for the next explicit or scheduled drain.
  • Queue reads and drains are ordered by SyncOperation.createdAt. When multiple parts of the same isolate enqueue work concurrently, provide deterministic createdAt values if strict cross-caller FIFO ordering matters.
  • drain(maxOperations: n) gives backpressure for large queues by processing a bounded batch and returning shouldContinue.
  • discardOperation, discardPendingForEntity, and the failed-operation discard helpers cover cancellation before an operation is sent.
  • Queue mutations are serialized with drain state transitions on an internal lock, so cleanup and retry APIs cannot race an in-flight send. Records being sent stay protected by their syncing status.
  • Priority scheduling is intentionally not part of 1.0.0. Per-entity ordering takes priority over arbitrary job priority.
  • A SyncEngine instance should be owned by one isolate. For multiple isolates, route queue writes through a single owner isolate or implement transactional locking in your SyncStore adapter.

For the full design boundary, see ARCHITECTURE.md.

UI State

engine.watchSyncState().listen((state) {
  if (state.isSyncing) {
    showSyncSpinner();
  }

  if (state.needsAttention) {
    showSyncIssueBadge();
  }
});

final taskRecords = await engine.readEntityRecords(
  const SyncEntityRef(type: 'task', id: 'task-1'),
);

final attentionRecords = await engine.readRecords(
  statuses: {SyncStatus.failed, SyncStatus.conflicted},
);

Use the Flutter builders when a widget should rebuild from sync state:

SyncStateBuilder(
  engine: engine,
  builder: (context, state, child) {
    if (state.isSyncing) {
      return const Text('Syncing');
    }

    if (state.needsAttention) {
      return const Text('Sync issue');
    }

    return const SizedBox.shrink();
  },
);

SyncEntityStateBuilder(
  engine: engine,
  entity: const SyncEntityRef(type: 'task', id: 'task-1'),
  builder: (context, state, child) {
    return Text(state.status.name);
  },
);

Use the prebuilt badges for compact status indicators:

SyncQueueStatusBadge(
  engine: engine,
  hideWhenIdle: true,
);

SyncEntityStatusBadge(
  engine: engine,
  entity: const SyncEntityRef(type: 'task', id: 'task-1'),
  hideWhenSynced: true,
);

SyncQueueDebugView(
  engine: engine,
  statuses: const {SyncStatus.failed, SyncStatus.conflicted},
);

Failures and Conflicts

Transport adapters detect protocol-level conflicts and return SyncResult.conflict. The engine persists that conflict and blocks newer work for the same entity, but the application decides whether to merge, discard, or fail the operation.

await engine.retryFailedOperations(
  entity: const SyncEntityRef(type: 'task', id: 'task-1'),
);

await engine.discardFailedOperations(
  entity: const SyncEntityRef(type: 'task', id: 'task-1'),
);

await engine.discardWhere(
  (record) => record.operation.entity.type == 'draft',
);

await engine.resolveConflict(
  'operation-1',
  const SyncConflictResolution.retry(
    payload: {'title': 'Merged title'},
  ),
);

See doc/conflict_resolution.md for a complete workflow.

Roadmap

  • Infrastructure-level queue replacement and compaction APIs.
  • More storage adapter packages beyond sync_queue_drift, such as Hive.
  • Stronger multi-isolate and multi-writer storage guidance.
  • More prebuilt Flutter diagnostics for retry schedules and conflicts.

Publishing

See PUBLISHING.md for the release validation and pub.dev publishing checklist.

Libraries

sync_queue
A local-first sync queue for durable Flutter mutations.