onSnapshotStateChange method

@override void Function()? onSnapshotStateChange(String snapshotId, void callback(SessionSnapshot snapshot), { Map<String, dynamic>? context, })

Watches a single snapshot file for changes and invokes callback with the parsed snapshot whenever it changes.

Unlike InMemorySessionStore, file-backed snapshots are frequently mutated by a different process (e.g. the request handler that received an abort writes status: 'aborted', while a detached background worker is the one watching). Detecting that requires observing the filesystem rather than in-process saveSnapshot calls.

Reliability comes from two layers:

• A directory watcher (Directory.watch) filtered to the target <snapshotId>.json. This is low latency but can miss events on some filesystems (network mounts, certain container volumes).
• A polling fallback (snapshotWatchPollInterval) that re-reads the file on an interval, backstopping any events the watcher drops.

Callbacks are de-duplicated by serialized content, so the noisy/duplicate events the watcher emits collapse into one callback per real change. Transient read errors (e.g. a partially written file mid-rewrite, or a not-yet-created file) are swallowed; the next event/poll re-reads.

Returns an unsubscribe function that stops watching and polling.

Implementation

@override
void Function()? onSnapshotStateChange(
  String snapshotId,
  void Function(SessionSnapshot snapshot) callback, {
  Map<String, dynamic>? context,
}) {
  _assertSafeSnapshotId(snapshotId);
  final dir = _prefixDir(context)..createSync(recursive: true);
  final fileName = '$snapshotId.json';
  final file = File('${dir.path}${Platform.pathSeparator}$fileName');

  var closed = false;
  var isReading = false;
  var needsRecheck = false;
  String? lastSerialized;

  // Re-read the file and fire the callback only when the content actually
  // changed. Watchers fire multiple events per write, so dedupe by content.
  // A non-overlapping loop (isReading / needsRecheck) ensures concurrent
  // events/poll ticks process sequentially without dropping a change.
  Future<void> emitIfChanged() async {
    if (closed) return;
    if (isReading) {
      needsRecheck = true;
      return;
    }
    isReading = true;
    try {
      do {
        needsRecheck = false;
        String contents;
        try {
          contents = await file.readAsString();
        } catch (_) {
          // Missing file (not yet created) or a transient read error during a
          // concurrent rewrite: ignore and wait for the next event/poll.
          continue;
        }
        if (closed || contents == lastSerialized) continue;
        SessionSnapshot snapshot;
        try {
          snapshot = SessionSnapshot.fromJson(
            jsonDecode(contents) as Map<String, dynamic>,
          );
        } catch (_) {
          // Partially written file mid-rewrite: skip without updating
          // lastSerialized so the next event/poll re-reads the complete file.
          continue;
        }
        lastSerialized = contents;
        callback(snapshot);
      } while (needsRecheck && !closed);
    } finally {
      isReading = false;
    }
  }

  // Watch the directory (not the file) so this still works before the file
  // exists and survives atomic rename-replace writes that swap the inode.
  StreamSubscription<FileSystemEvent>? watcher;
  try {
    watcher = dir.watch().listen((event) {
      if (event.path == file.path) unawaited(emitIfChanged());
    });
  } catch (_) {
    // Some environments disallow directory watching; polling covers us.
  }

  final pollTimer = Timer.periodic(
    _snapshotWatchPollInterval,
    (_) => unawaited(emitIfChanged()),
  );

  // Surface the current state immediately (if the file already exists).
  unawaited(emitIfChanged());

  return () {
    closed = true;
    unawaited(watcher?.cancel());
    pollTimer.cancel();
  };
}