Temporal Patterns in Production - Sagas, Child Workflows, Continue-As-New, Versioning & Worker Scaling

"""Two production patterns inside a durable workflow, simulated.

Part 1, saga with compensation (orchestrated saga):
  - transient failures are retried by the activity retry policy (with backoff)
  - a non-retryable business failure triggers compensations in reverse order
  - the compensation is registered BEFORE calling the step, because a step can
    succeed on the remote side and still look failed to us (timeout)
  - compensations are idempotent and retried until they succeed

Part 2, continue-as-new for a long-lived entity workflow:
  - every handled event grows the event history
  - before a threshold, the workflow checkpoints its state and continues as a
    new run (same Workflow ID, new Run ID, fresh history)
"""
import itertools

class Transient(Exception): pass
class NonRetryable(Exception): pass

attempts = {}
remote_bookings = set()                       # what really exists at the providers (a set: providers dedup on an idempotency key)

def provider(step):
    """The outside world, scripted. Returns normally or raises."""
    n = attempts[step] = attempts.get(step, 0) + 1
    if step == "book_hotel":
        remote_bookings.add("hotel")          # booking succeeds remotely...
        if n == 1: raise Transient("timeout") # ...but our first call times out
    if step == "book_flight": remote_bookings.add("flight")
    if step == "book_car": raise NonRetryable("NoCarsAvailable")
    if step == "cancel_flight" and n == 1: raise Transient("503 from airline")
    if step.startswith("cancel_"): remote_bookings.discard(step[len("cancel_"):])
    return f"{step}:ok"

def activity(step, max_attempts=5, initial=1.0, coeff=2.0):
    """Retry policy: exponential backoff, stop on non-retryable errors."""
    delay = initial
    for a in itertools.count(1):
        try:
            r = provider(step); print(f"    {step:<14} attempt {a}: ok"); return r
        except NonRetryable as e:
            print(f"    {step:<14} attempt {a}: NON-RETRYABLE {e}"); raise
        except Transient as e:
            if a >= max_attempts: raise
            print(f"    {step:<14} attempt {a}: transient ({e}), retry in {delay:.0f}s"); delay *= coeff

def trip_saga():
    compensations = []
    try:
        for step, undo in [("book_flight", "cancel_flight"), ("book_hotel", "cancel_hotel"), ("book_car", "cancel_car")]:
            compensations.append(undo)        # register first: the step may half-succeed
            activity(step)
        return "booked"
    except NonRetryable:
        print("    -> compensating in reverse order")
        for undo in reversed(compensations):
            activity(undo, max_attempts=100)  # compensations must eventually succeed
        return "rolled back"

print("Part 1: trip booking saga")
print("  result:", trip_saga(), "| still booked at providers:", sorted(remote_bookings) or "nothing")

print("\nPart 2: entity workflow with continue-as-new (demo threshold: 20 events)")
EVENTS_PER_SIGNAL = 3          # e.g. signal received + workflow task scheduled/completed (simplified)
THRESHOLD = 20                 # real Temporal warns at 10,240 events and hard-limits at 51,200 events or 50 MB

def account_run(run_id, state, deposits):
    history = 2                # started + first workflow task
    while deposits:
        amt = deposits.pop(0); state["balance"] += amt; state["seq"] += 1; history += EVENTS_PER_SIGNAL
        if history + EVENTS_PER_SIGNAL > THRESHOLD and deposits:
            print(f"  run {run_id}: {history} events, balance={state['balance']} seq={state['seq']} -> continue-as-new")
            return state, True
    print(f"  run {run_id}: {history} events, balance={state['balance']} seq={state['seq']} -> idle, waiting for signals")
    return state, False

deposits = [10] * 14
state, run = {"balance": 0, "seq": 0}, 1
while True:
    state, cont = account_run(f"account-7/run-{run}", state, deposits)
    if not cont: break
    run += 1
print(f"  same Workflow ID 'account-7', {run} runs, no history ever exceeded {THRESHOLD} events")

Part 1: trip booking saga
    book_flight    attempt 1: ok
    book_hotel     attempt 1: transient (timeout), retry in 1s
    book_hotel     attempt 2: ok
    book_car       attempt 1: NON-RETRYABLE NoCarsAvailable
    -> compensating in reverse order
    cancel_car     attempt 1: ok
    cancel_hotel   attempt 1: ok
    cancel_flight  attempt 1: transient (503 from airline), retry in 1s
    cancel_flight  attempt 2: ok
  result: rolled back | still booked at providers: nothing

Part 2: entity workflow with continue-as-new (demo threshold: 20 events)
  run account-7/run-1: 20 events, balance=60 seq=6 -> continue-as-new
  run account-7/run-2: 20 events, balance=120 seq=12 -> continue-as-new
  run account-7/run-3: 8 events, balance=140 seq=14 -> idle, waiting for signals
  same Workflow ID 'account-7', 3 runs, no history ever exceeded 20 events

// Changing workflow code while executions are in flight.
// v1: charge -> ship.  v2 adds a fraud check before charge.
// Replaying an old (v1) history with v2 code breaks determinism unless the change
// is guarded by a patch marker (Temporal: patched()/GetVersion), or the old run is
// pinned to the old build (Worker Versioning). Simplified simulation.

type Hist = { cmds: string[]; markers: Set<string> };
type Code = (ctx: Ctx) => string[];
class NonDeterminism extends Error {}

class Ctx {
  private i = 0; issued: string[] = [];
  constructor(private h: Hist, private replaying: boolean) {}
  patched(id: string): boolean {
    // Not replaying: record a marker and take the new path.
    // Replaying: take the new path only if the original run recorded the marker.
    if (!this.replaying) { this.h.markers.add(id); return true; }
    return this.h.markers.has(id);
  }
  activity(name: string) {
    if (this.replaying && this.i < this.h.cmds.length && this.h.cmds[this.i] !== name)
      throw new NonDeterminism(`history step ${this.i + 1} is ${this.h.cmds[this.i]}, code asked for ${name}`);
    if (this.i >= this.h.cmds.length) this.h.cmds.push(name);
    this.i++; this.issued.push(name);
  }
}

const v1: Code = (c) => { c.activity("charge"); c.activity("ship"); return c.issued; };
const v2Unsafe: Code = (c) => { c.activity("fraud_check"); c.activity("charge"); c.activity("ship"); return c.issued; };
const v2Patched: Code = (c) => {
  if (c.patched("add-fraud-check")) c.activity("fraud_check");
  c.activity("charge"); c.activity("ship"); return c.issued;
};

// An order that started on v1 and is mid-flight: 'charge' is already in its history.
const inflightV1 = (): Hist => ({ cmds: ["charge"], markers: new Set() });

function tryReplay(label: string, code: Code, h: Hist) {
  try { console.log(`  ${label}: ok, commands = ${code(new Ctx(h, true)).join(" > ")}`); }
  catch (e) { console.log(`  ${label}: ${(e as Error).constructor.name}: ${(e as Error).message}`); }
}

console.log("1) Deploy v2 and replay an in-flight v1 order");
tryReplay("v2 unsafe ", v2Unsafe, inflightV1());
tryReplay("v2 patched", v2Patched, inflightV1());

console.log("2) A brand-new order on v2 patched code records the marker");
const fresh: Hist = { cmds: [], markers: new Set() };
console.log(`  new run: ${v2Patched(new Ctx(fresh, false)).join(" > ")}, markers = [${[...fresh.markers]}]`);
tryReplay("replay of that new run", v2Patched, fresh);

console.log("3) Patch lifecycle: patched() -> deprecate the patch -> delete it");
console.log("   only after no open (or retained, replayable) execution still lacks the marker");

console.log("4) Worker Versioning alternative: route by build instead of branching in code");
const current = "build-v2";
for (const wf of [{ id: "order-1", startedOn: "build-v1", behavior: "Pinned" }, { id: "order-2", startedOn: "build-v1", behavior: "AutoUpgrade" }, { id: "order-3", startedOn: "build-v2", behavior: "Pinned" }]) {
  const target = wf.behavior === "Pinned" ? wf.startedOn : current;
  const note = wf.behavior === "AutoUpgrade" && wf.startedOn !== current ? " (must stay replay-safe: still needs patching)" : "";
  console.log(`   ${wf.id} started on ${wf.startedOn}, ${wf.behavior} -> next workflow task goes to ${target}${note}`);
}

1) Deploy v2 and replay an in-flight v1 order
  v2 unsafe : NonDeterminism: history step 1 is charge, code asked for fraud_check
  v2 patched: ok, commands = charge > ship
2) A brand-new order on v2 patched code records the marker
  new run: fraud_check > charge > ship, markers = [add-fraud-check]
  replay of that new run: ok, commands = fraud_check > charge > ship
3) Patch lifecycle: patched() -> deprecate the patch -> delete it
   only after no open (or retained, replayable) execution still lacks the marker
4) Worker Versioning alternative: route by build instead of branching in code
   order-1 started on build-v1, Pinned -> next workflow task goes to build-v1
   order-2 started on build-v1, AutoUpgrade -> next workflow task goes to build-v2 (must stay replay-safe: still needs patching)
   order-3 started on build-v2, Pinned -> next workflow task goes to build-v2