tick method

Progress tick()

Run one loop pass. See the class docs for the resync → drain → retain → receive shape. Sink failures retain-and-stall (not an error); only an inbound source read failure throws DriverError.

Implementation

Progress tick() {
  final now = _clock.nowMillis();
  final progress = Progress();

  // 1. resync-on-reconnect: replay the unacked outbox suffix, oldest first.
  if (_replayPending) {
    _replayPending = false;
    for (final (_, msg) in _outbox.replayFrom(_peerAckedThrough)) {
      if (_sink.send(msg)) {
        progress.sent += 1;
      } else {
        _stalledSince = now;
        _replayPending = true; // finish the replay after the next reconnect
        break;
      }
    }
  }

  // 2. drain fresh enqueues: append-before-send, retain-and-stop on failure.
  //    A pre-existing stall (a prior failed send, no reconnect yet) skips the
  //    drain entirely — do not push into a sink already known to be down.
  while (_stalledSince == null) {
    if (_pending.isEmpty) break;
    final (epoch, msg) = _pending.first;
    _outbox.append(epoch, msg);
    _pending.removeFirst();
    if (_sink.send(msg)) {
      progress.sent += 1;
      _stalledSince = null;
    } else {
      // Retained in the outbox (unacked) → replayed on reconnect.
      _stalledSince = now;
      break;
    }
  }

  // 3. receive: route control frames + feed data frames through the
  //    coordinator.
  while (true) {
    IpcMessage? msg;
    try {
      msg = _source.recv();
    } on DriverError {
      rethrow;
    } catch (e) {
      throw DriverError.source(e);
    }
    if (msg == null) break;
    switch (msg) {
      case IpcMessageOutboxAck(:final value):
        if (value.throughEpoch > _peerAckedThrough) {
          _peerAckedThrough = value.throughEpoch;
        }
        _outbox.ackThrough(value.throughEpoch);
      case IpcMessageResyncRequest(:final value):
        final snap = _provider.snapshot(value.fromEpoch);
        if (_sink.send(snap)) {
          progress.snapshotsServed += 1;
        } else {
          _stalledSince = now;
        }
      case IpcMessageCrdtSync():
        // Idempotent anti-entropy plane — the host folds it directly.
        progress.applied.add(msg);
      case IpcMessageSnapshot() || IpcMessageDelta():
        final action = _coordinator.ingest(msg);
        switch (action) {
          case ResyncActionApply():
            _ackOwed = true;
            progress.applied.add(msg);
          case ResyncActionRequestSnapshot(:final fromEpoch):
            final req = IpcMessage.ofResyncRequest(
                ResyncRequest(fromEpoch: fromEpoch));
            if (_sink.send(req)) {
              progress.resyncRequested = true;
            } else {
              _stalledSince = now;
            }
          case ResyncActionIgnore():
            break;
        }
    }
  }

  // 4. advertise our receiver cursor if we applied anything (retry until
  //    sent).
  if (_ackOwed && _sink.send(_coordinator.ack())) {
    _ackOwed = false;
  }

  progress.peerAckedThrough = _peerAckedThrough;
  progress.retained = _outbox.retainedEpochs().length;
  return progress;
}