Fenced writes with FoldWrite

There are two ways to process a batch of a key's inputs:

• Hand it to the store's own group operation. The batch call passes the batch through; no helper is needed.
• Read once, apply in memory, write once. Read the key's state, apply the inputs to it in order, and write the result only if the store still holds what was read. FoldWrite packages this round trip.

The round trip

FoldWrite(id, read, step, write, retry):

Parameter	What it does
id: I => Id	Identifies an input. Inputs of one batch with the same id are applied once and share the first one's outcome.
read: K => ZIO[R, E, S]	Reads a key's state, including whatever makes the write conditional, such as a version or a length.
step: (S, I) => (S, O)	Your logic: the next state and the input's outcome. It must be pure: it runs again on every retry.
write: (K, S, S) => ZIO[R, E, Any]	Writes after if the store still holds before, and fails otherwise. Skipped when nothing changed.
retry: Schedule[R, E, Any]	When to retry the whole round trip after read or write failed.

fold.call is the batch call and fold.outcome the extract:

/**
 * One round trip per batch: read the log, apply the entries, and append what is new only if the log has not changed
 * since it was read. Only a store error fails, and retries, the whole batch.
 */
def write(store: Store): FoldWrite[Any, StoreError, String, Entry, String, Vector[String], Either[String, Int]] =
  FoldWrite(
    id = _.id,
    read = store.read,
    step = step,
    write = (entity, before, after) => store.append(entity, before.size, Chunk.fromIterable(after.drop(before.size))),
    retry = Schedule.exponential(10.millis).jittered && Schedule.recurs(5)
  )

Making retries safe

A write can land and its response be lost. The retry then reads a state that already contains the batch's inputs. step must answer such an input from the state, without applying it again:

/**
 * Applies one entry to an entity's log as read: an invalid entry is rejected without failing the others, and an entry
 * already in the log (a retried write may have landed) is answered with its position instead of being appended again.
 */
def step(log: Vector[String], entry: Entry): (Vector[String], Either[String, Int]) =
  if entry.amount == 0 then log -> Left("amount must not be zero")
  else if log.contains(entry.id) then log -> Right(log.indexOf(entry.id))
  else (log :+ entry.id)                  -> Right(log.size)

Here an invalid input gets its own rejection and fails no other input. An input already in the log is answered with its position. Only a store error fails the batch, and that is the case retry retries.

To answer a retried input with exactly the answer it got the first time, keep the answers in the state: S holds, for each applied id, what step returned, and step returns it again instead of applying the input.

Each input is answered as of its own position

step runs once per input, in order, on the state left by the inputs before it. An input's outcome is computed then, from that state, not from the state the batch ends with: in a batch of three appends to an empty log, the first is answered with position 0, the second with 1 and the third with 2.

Order that lives in the state

Some inputs change how the ones after them apply: closing a period, rolling a business date over, switching a schedule. That is a state transition, and it belongs in step:

final case class Ledger(period: Int, totals: Map[Int, Long])

enum Posting:
  case Add(amount: Long)
  case Close

/** Closing a period moves every later posting of the batch, and of later batches, to the next one. */
def post(ledger: Ledger, posting: Posting): (Ledger, Int) = posting match
  case Posting.Add(amount) =>
    val total = ledger.totals.getOrElse(ledger.period, 0L) + amount
    ledger.copy(totals = ledger.totals.updated(ledger.period, total)) -> ledger.period
  case Posting.Close => ledger.copy(period = ledger.period + 1) -> (ledger.period + 1)

In the batch Add(1), Close, Add(2) the first amount lands in period 0 and the second in period 1. Order.by can make such inputs meet in the right order when they wait together; for an input that arrives after a later one was already written, the fold over the state decides, as for any late input (0011).

Inputs that depend on the ones before them

By default each input is judged on its own: a rejected input leaves the state as it was, and the inputs after it in the batch apply as if it had not been there. One caller's invalid input never fails another caller's valid one (0006).

Some keys are not like that. If an account's entries must apply in sequence, an entry after a rejected one must not be attempted. That rule is about the key, not about the batch: which inputs share a batch is a matter of timing, so a rule that only looked back inside one batch would answer the same inputs differently under load. Put the block in the state instead, where step sees it for every later input, in this batch and the next:

/** An account's balance, and the id of the entry that blocked it, if one did. */
final case class Account(balance: Long, blockedBy: Option[String])

final case class Debit(id: String, amount: Long)

enum Rejection:
  case InsufficientFunds
  case NotAttempted(blockedBy: String)

/**
 * An account's entries depend on the ones before them: once one is rejected, the ones after it are not attempted, in
 * this batch or in a later one, until something unblocks the account. The block is part of the state, so it is
 * written with the batch and survives the batch boundary, a retry and the process.
 */
def debit(account: Account, entry: Debit): (Account, Either[Rejection, Long]) =
  account.blockedBy match
    case Some(id) if id != entry.id =>
      account -> Left(Rejection.NotAttempted(id)) // the blocking entry, replayed, is judged again
    case _ if entry.amount > account.balance =>
      account.copy(blockedBy = Some(entry.id)) -> Left(Rejection.InsufficientFunds)
    case _ =>
      val balance = account.balance - entry.amount
      account.copy(balance = balance) -> Right(balance)

A caller then gets one of three answers: its result, its own rejection, or NotAttempted naming the entry that blocked the key. Lifting the block is an entry of its own, a correction or a reversal, which step applies by clearing blockedBy; until it arrives, every entry of the key is answered NotAttempted without a write. A rejected entry that blocked the key does change the state, so the batch is written, and a retry after a lost response finds the block in place and answers the same way.

If the dependency really should last only for the batch, keep the block in a wrapper around the state that read creates unblocked and write strips; FoldWrite writes whenever the wrapper changed, so a batch that only set the block costs one write of an unchanged record.

step must be total

step returns an outcome; it must not throw. An exception in step is a defect of the whole call, and every caller in the batch sees it. A rejection, a "not attempted", a validation error are all values of O.

Putting it together

config <- ZIO.fromEither(
  // No delay: an entity's entries wait only while its previous write is in flight, and a quiet entity's entry is
  // written at once. A small positive delay would let more out-of-order entries meet and be sorted, at that
  // much latency. The weight limit keeps a write under the store's request size limit, in (estimated) bytes.
  PerKeyConfig.make(maxBatchSize = 25, maxInFlightBatches = 8, maxQueueing = 1000, maxBatchWeight = 2000)
)
fold = write(store)
writer <- Ferry.perKey(
  config,
  (_: Entry).entity,
  (e: Entry) => 40L + e.id.length,
  Order.by((_: Entry).effectiveDay)
)(fold.call)(fold.outcome)

The OrderedWrites example runs this against a store that throttles one request in seven. It writes 300 entries in about twenty requests, each entry exactly once, ordered by effective day.

Why not cache the state?

FoldWrite reads the state for every batch. Keeping it in memory between batches would save a read, but make the process a second source of truth: wrong after a crash, after another process wrote the key, or after a routing change. The store stays the only source of truth. See 0005 and Running on a fleet.

Two writes: commit, then project

A store that writes a batch in two steps, a commit and then a projection into a read-side view, needs a second fenced write, a way to report "committed, view pending", and a way to finish the projection after a crash. ProjectedWrite wraps a FoldWrite and adds all three.

All-or-nothing group operations

If the store's group operation rejects a whole group for one invalid input, validate in call before the round trip, or split and retry inside the call. The key stays busy meanwhile, so order is preserved.