Persistence & Postgres Adapter

Invariant decouples execution semantics from physical storage. All durability guarantees are exposed through clean TypeScript interfaces in @invariant/core, with @invariant/postgres serving as the official, production-grade storage adapter.

The 4 Core Storage Interfaces

A storage adapter in Invariant implements four distinct contracts:

// 1. Event Log Storage
export interface EventStore {
  append(runId: string, event: UnstoredRuntimeEvent): Promise<StoredRuntimeEvent>;
  readLog(runId: string): Promise<StoredRuntimeEvent[]>;
}

// 2. State & Revision Management
export interface StateStore {
  loadState(runId: string): Promise<ExecutionState>;
  commitState(runId: string, expectedRevision: number, state: ExecutionState): Promise<void>;
}

// 3. Worker Concurrency & Leases
export interface LeaseStore {
  acquire(runId: string, workerId: string, ttlMs: number): Promise<Lease | null>;
  renew(runId: string, leaseId: string, ttlMs: number): Promise<boolean>;
  release(runId: string, leaseId: string): Promise<void>;
}

// 4. Transactional Outbox
export interface OutboxStore {
  enqueue(command: ExecutionCommand): Promise<void>;
  pending(limit: number): Promise<StoredCommand[]>;
  complete(commandId: string, resultEvent: UnstoredRuntimeEvent): Promise<void>;
}

The Atomic Transaction Contract (RuntimeTransaction)

While storage interfaces are segregated for clean reading and individual lookups, a transition commit requires guaranteed atomicity.

Any storage adapter MUST implement RuntimeTransaction:

export interface RuntimeTransaction {
  commitTransition(input: {
    runId: string;
    expectedRevision: number;
    state: ExecutionState;
    events: UnstoredRuntimeEvent[];
    commands: ExecutionCommand[];
  }): Promise<void>;
}

PostgreSQL Implementation (PostgresRuntimeStore)

Under the hood, @invariant/postgres satisfies RuntimeTransaction using PostgreSQL row-level locks and transactional Outbox inserts:

export class PostgresRuntimeStore implements RuntimeTransaction {
  constructor(private pool: pg.Pool) {}

  async commitTransition({ runId, expectedRevision, state, events, commands }) {
    const client = await this.pool.connect();
    try {
      await client.query("BEGIN");

      // 1. Lock run & verify optimistic revision
      const runRes = await client.query(
        "SELECT revision FROM workflow_runs WHERE id = $1 FOR UPDATE",
        [runId]
      );
      if (runRes.rows[0].revision !== expectedRevision) {
        throw new WorkflowRevisionConflictError(runId, expectedRevision);
      }

      // 2. Append events sequentially
      for (const event of events) {
        await client.query(
          "INSERT INTO workflow_events (run_id, type, payload) VALUES ($1, $2, $3)",
          [runId, event.type, event.payload]
        );
      }

      // 3. Update state and increment revision
      await client.query(
        "UPDATE workflow_states SET state = $1, revision = revision + 1 WHERE run_id = $2",
        [JSON.stringify(state), runId]
      );

      // 4. Insert Outbox side-effects
      for (const cmd of commands) {
        await client.query(
          "INSERT INTO workflow_outbox (run_id, command_type, payload) VALUES ($1, $2, $3)",
          [runId, cmd.type, cmd.payload]
        );
      }

      await client.query("COMMIT");
    } catch (err) {
      await client.query("ROLLBACK");
      throw err;
    } finally {
      client.release();
    }
  }
}

Rules for Custom Storage Adapters

If you write a custom adapter (e.g. for MySQL, SQLite, or CockroachDB), it MUST NOT break these invariants:

1. ❌ No Non-Atomic Splits: You cannot commit state in one query and outbox in an un-persisted async background call.
2. ❌ No Dirty Reads: loadState() must return the state representing the latest committed revision.
3. ❌ No Lease Overlaps: acquire() must enforce mutual exclusion—only one worker can hold an active lease on a runId at a time.