Distributed transactions and sagas

// Commit runs work on every participant as one global transaction named txid.
func (c *Coordinator) Commit(ctx context.Context, txid string, work map[string]Work) error {
  for i, p := range c.Participants { // phase 1: every participant prepares, or nobody commits
    if err := Prepare(ctx, p, txid, work[p.Name]); err != nil {
      return errors.Join(fmt.Errorf("%s voted no: %w", p.Name, err), c.abort(ctx, txid, c.Participants[:i]))
    }
    c.step(p.Name, "PREPARE TRANSACTION '"+p.GID(txid)+"'")
  }
  if c.CrashAt == "before decision" {
    return ErrCrashed // every participant waits; recovery finds no decision and rolls back
  }
  // The decision point. Once this row is durable, the transaction is committed.
  if _, err := c.Log.Exec(ctx, twopc["decide"], txid, "commit"); err != nil {
    return errors.Join(fmt.Errorf("log decision: %w", err), c.abort(ctx, txid, c.Participants))
  }
  c.step("coordinator", "log decision: commit")
  if c.CrashAt == "after decision" {
    return ErrCrashed // every participant waits, holding its locks
  }
  for _, p := range c.Participants { // phase 2: may not fail; Recover retries it
    if err := Finish(ctx, p.Conn, p.GID(txid), true); err != nil {
      return fmt.Errorf("commit on %s, retry with Recover: %w", p.Name, err)
    }
  }
  return nil
}

-- The coordinator's log. It must reach disk before any participant hears "commit".
CREATE TABLE twopc_decisions (
  gid      text PRIMARY KEY,
  decision text NOT NULL CHECK (decision IN ('commit', 'abort'))
);

-- Prepared transactions in this database: they hold their locks until someone decides.
SELECT gid FROM pg_prepared_xacts WHERE database = current_database() ORDER BY gid;

// PlaceOrder writes the order and its event in one local transaction. Both exist or neither does.
// The service never talks to the broker; the relay does.
func PlaceOrder(ctx context.Context, db *pgxpool.Pool, o Order) error {
  payload, err := eventFor(o)
  if err != nil {
    return err
  }
  return pgx.BeginFunc(ctx, db, func(tx pgx.Tx) error {
    if _, err := tx.Exec(ctx, ob["place_order"], o.ID, o.SKU, o.Qty, o.CardID, o.AmountCents, o.Region); err != nil {
      return fmt.Errorf("place order %d: %w", o.ID, err)
    }
    if _, err := tx.Exec(ctx, ob["write_event"], "orders", payload); err != nil {
      return fmt.Errorf("write event for order %d: %w", o.ID, err)
    }
    return nil
  })
}

// PublishBatch publishes up to n of the oldest unpublished events and returns how many it sent.
// Delivery is at least once: a crash before the commit sends the same events again later.
func (r *Relay) PublishBatch(ctx context.Context, n int) (sent int, err error) {
  err = pgx.BeginFunc(ctx, r.DB, func(tx pgx.Tx) error {
    rows, err := tx.Query(ctx, ob["next_batch"], n)
    if err != nil {
      return fmt.Errorf("read outbox: %w", err)
    }
    var id int64
    var payload string
    var ids []int64
    pipe := r.RDB.Pipeline()
    if _, err := pgx.ForEachRow(rows, []any{&id, &payload}, func() error {
      ids = append(ids, id)
      pipe.XAdd(ctx, &redis.XAddArgs{Stream: r.Stream, Values: []any{"msg_id", id, "payload", payload}})
      return nil
    }); err != nil {
      return fmt.Errorf("read outbox: %w", err)
    }
    if len(ids) == 0 {
      return nil
    }
    if _, err := pipe.Exec(ctx); err != nil {
      return fmt.Errorf("publish %d events: %w", len(ids), err)
    }
    if r.CrashAfterPublish {
      r.CrashAfterPublish = false
      return ErrRelayCrashed // the transaction rolls back: the rows stay unpublished
    }
    if _, err := tx.Exec(ctx, ob["mark_published"], ids); err != nil {
      return fmt.Errorf("mark published: %w", err)
    }
    sent = len(ids)
    return nil
  })
  return sent, err
}

// handle applies one message. The inbox row and the effect commit together, so a message id is
// applied at most once, however many times the broker delivers it.
func (c *Consumer) handle(ctx context.Context, m redis.XMessage) (applied bool, err error) {
  msgID, err := strconv.ParseInt(fmt.Sprint(m.Values["msg_id"]), 10, 64)
  if err != nil {
    return false, fmt.Errorf("message %s: bad msg_id: %w", m.ID, err)
  }
  var ev orderPlaced
  if err := json.Unmarshal([]byte(fmt.Sprint(m.Values["payload"])), &ev); err != nil {
    return false, fmt.Errorf("message %s: bad payload: %w", m.ID, err)
  }
  err = pgx.BeginFunc(ctx, c.DB, func(tx pgx.Tx) error {
    if c.Inbox {
      tag, err := tx.Exec(ctx, ob["inbox_claim"], msgID)
      if err != nil {
        return fmt.Errorf("claim message %d: %w", msgID, err)
      }
      if tag.RowsAffected() == 0 {
        return nil // seen before: skip the effect
      }
    }
    if _, err := tx.Exec(ctx, ob["send_email"], ev.OrderID); err != nil {
      return fmt.Errorf("send email for order %d: %w", ev.OrderID, err)
    }
    applied = true
    return nil
  })
  return applied, err
}

-- The relay takes the oldest unpublished events. SKIP LOCKED lets several relays share the work.
SELECT id, payload::text FROM outbox
WHERE published_at IS NULL
ORDER BY id
LIMIT $1
FOR UPDATE SKIP LOCKED;

UPDATE outbox SET published_at = now() WHERE id = ANY($1);

-- The consumer records the message id in the same transaction as the effect.
INSERT INTO inbox (message_id) VALUES ($1) ON CONFLICT (message_id) DO NOTHING;

-- The effect. Without the inbox, a redelivery runs it twice: sent becomes 2.
INSERT INTO order_emails (order_id) VALUES ($1)
ON CONFLICT (order_id) DO UPDATE SET sent = order_emails.sent + 1;