commitJournal method

  1. @override
Future<bool> commitJournal(
  1. WorkflowJournalEntry entry, {
  2. required int expectedRevision,
  3. required String executionId,
  4. WorkflowJournalCheckpoint? checkpoint,
})
override

Commits a revision and optional checkpoint, returning false on conflict.

Throws ArgumentError for invalid write arguments, as defined by WorkflowJournalEntry.validateWrite. These are caller errors, not optimistic-concurrency conflicts, and must not be retried as contention.

WorkflowJournalEntry.revision must equal expectedRevision + 1. Missing records have revision zero. Forward writes require a running run; compensation writes require a failed run. Both require the exact executionId. A compensation registration gets a unique monotonically increasing completion position.

Ordinary administrative rewind must delete journal records for discarded checkpoints and invalidate existing execution/compensation claims.

Implementation

@override
Future<bool> commitJournal(
  WorkflowJournalEntry entry, {
  required int expectedRevision,
  required String executionId,
  WorkflowJournalCheckpoint? checkpoint,
}) async {
  entry.validateWrite(
    expectedRevision: expectedRevision,
    checkpoint: checkpoint,
  );
  final run = _runs[entry.runId];
  final requiredStatus = entry.kind == WorkflowJournalKind.step
      ? WorkflowStatus.running
      : WorkflowStatus.failed;
  if (run == null ||
      run.status != requiredStatus ||
      run.executionId != executionId ||
      executionId.isEmpty) {
    return false;
  }
  final key = (entry.kind, entry.name);
  final records = _journal[entry.runId];
  final previous = records?[key];
  if ((previous?.revision ?? 0) != expectedRevision ||
      (entry.kind == WorkflowJournalKind.compensation && previous == null)) {
    return false;
  }
  // Validate/copy before publishing any part of the atomic mutation.
  final data = _copyJournalValue(entry.data)! as Map<String, Object?>;
  final value = _copyJournalValue(checkpoint?.value);
  final registration = checkpoint?.compensation;
  final registrationData = registration == null
      ? null
      : _copyJournalValue(registration.toJournalData())!
            as Map<String, Object?>;
  final target = _journal.putIfAbsent(entry.runId, () => {});
  target[key] = WorkflowJournalEntry(
    runId: entry.runId,
    kind: entry.kind,
    name: entry.name,
    revision: entry.revision,
    data: data,
    position: previous?.position,
  );
  if (checkpoint != null) {
    _steps.putIfAbsent(entry.runId, () => {})[entry.name] = value;
    final compensationKey = (WorkflowJournalKind.compensation, entry.name);
    if (registrationData != null && !target.containsKey(compensationKey)) {
      final position = (_journalOrder[entry.runId] ?? 0) + 1;
      _journalOrder[entry.runId] = position;
      target[compensationKey] = WorkflowJournalEntry(
        runId: entry.runId,
        kind: WorkflowJournalKind.compensation,
        name: entry.name,
        revision: 1,
        position: position,
        data: registrationData,
      );
    }
  }
  _runs[entry.runId] = run.copyWith(updatedAt: _clock.now());
  return true;
}