micro--go-micro
Asim Aslam
9fdcc24cce
* docs: design note for flow steps + Checkpoint durable execution
* docs: fold in durable-execution decisions (State struct, single Step, run retention, retry)
* docs: rename State.Payload to State.Data
* feat(flow): ordered steps + Checkpoint durable execution
A flow can now be an ordered list of steps (a task with stages) instead
of a single LLM turn. State carries typed Data plus a Stage marker; each
step is checkpointed before and after via a pluggable Checkpoint
(store-backed by default), so a run survives a crash and resumes where it
stopped without re-running completed steps. Flow-level Retry with a
per-step override; runs retained for audit unless DeleteOnSuccess.
Step actions: Call (RPC), LLM (augmented turn), Dispatch (to an agent),
or any StepFunc. Single-step and agent-dispatch flows are unchanged.
* feat(flow): top-level re-exports + durable flow example
Expose the step/checkpoint API from the micro package (FlowSteps,
FlowStep, FlowState, FlowRetry, FlowWithCheckpoint, FlowCall/LLM/Dispatch,
Checkpoint, StoreCheckpoint) and add a runnable, key-free example
demonstrating crash + resume.
* docs: document durable flow steps (guide, README, CLI help)
* docs: blog post + changelog for durable workflows
* fix(flow): scope checkpoint keys by flow name (flow/{name}/runs/{id})
Run keys were flow/runs/{id} — a single global keyspace shared by every
flow on the default store. Namespace them by flow name so each flow's
state is kept apart. StoreCheckpoint now takes a scope argument (the flow
passes its name by default).
* feat(store): Scope handle; scope agent and flow state by name
Add store.Scope(s, database, table) — a store handle that confines every
operation to a database/table without mutating the shared store, so
co-located components don't clobber each other's table (the failure mode
of the global Init(Table(...)) approach).
Use it to keep each agent's memory and plan in its own table
(agent/{name}) and each flow's runs in its own (flow/{name}), instead of
one global table partitioned only by key prefix. Services already scope
by service name.
* feat: consistent state model — service store scoping, flow registry, list/history CLI
- service: scope store via store.Scope (database service / table name),
retiring the Init(store.Table(name)) global-mutation hack; bridge the
default store so handlers using store.DefaultStore stay isolated.
- flow: register in the registry as type=flow while running (with trigger
and step count), deregister on Stop. Live discovery, like agents.
- cli: micro flow list (registry), micro flow runs <name> (durable store),
micro agent history <name> (durable store). list = running, runs/history
= durable, mirroring the service model.
* test: mini-universe end-to-end harness + scheduled GitHub Action
internal/harness/universe boots a small but real go-micro world — four
services, a durable checkout flow that crashes at payment and resumes,
and a guardrailed agent with a tool wrapper reached over RPC — drives the
scenario, asserts the end state (10 checks), and shuts down. Everything
is real except the LLM (mocked), so it's deterministic and needs no key;
-provider anthropic runs it live. Exits non-zero on failure, so it's an
end-to-end test, not just a demo.
Adds .github/workflows/universe.yml (push/PR/daily/dispatch) running the
universe + existing harnesses on the mock provider, plus an opt-in job
that runs live when ANTHROPIC_API_KEY is set. 'make harness' runs them
locally.
* ci: run the live universe job against AtlasCloud (ATLASCLOUD_API_KEY)
* ci: run the live universe job only on schedule or manual dispatch
The deterministic mock job still runs on push/PR/daily; the live
(AtlasCloud) job runs daily and on manual workflow_dispatch only, so
changes don't burn API credits on every PR but can still be checked
against a real model on demand.
---------
Co-authored-by: Claude <noreply@anthropic.com>
73 行
2.5 KiB
Markdown
73 行
2.5 KiB
Markdown
# Durable Flow
|
|
|
|
A workflow that survives a crash and resumes where it stopped.
|
|
|
|
A `flow` can be an ordered list of **steps** — a task with stages —
|
|
instead of a single LLM turn. Each step is checkpointed before and after
|
|
through a pluggable `Checkpoint` (store-backed by default), so if the
|
|
process dies mid-run, the run resumes at the step it stopped on, without
|
|
re-running the steps that already completed (and already had their side
|
|
effects).
|
|
|
|
## What this shows
|
|
|
|
A three-step checkout (`reserve → charge → confirm`) whose `charge` step
|
|
fails the first time, simulating a transient outage / crash:
|
|
|
|
```
|
|
first run:
|
|
reserve → inventory reserved
|
|
charge → payment dependency unavailable (crash)
|
|
run failed: payment gateway timeout
|
|
|
|
checkpoint: run 70643f61 is at step "charge" (status failed)
|
|
|
|
resume:
|
|
charge → payment captured
|
|
confirm → order confirmed
|
|
|
|
reserve ran 1 time(s) total — completed steps are not repeated on resume
|
|
no pending runs — the workflow completed durably
|
|
```
|
|
|
|
The key line is the last pair: on `Resume`, `reserve` does **not** run
|
|
again — its result was checkpointed — and the run finishes.
|
|
|
|
## The pieces
|
|
|
|
```go
|
|
f := micro.NewFlow("checkout",
|
|
micro.FlowSteps(
|
|
micro.FlowStep{Name: "reserve", Run: reserve},
|
|
micro.FlowStep{Name: "charge", Run: charge},
|
|
micro.FlowStep{Name: "confirm", Run: confirm},
|
|
),
|
|
micro.FlowWithCheckpoint(micro.StoreCheckpoint(nil, "checkout")), // nil store = default; "checkout" = key scope
|
|
)
|
|
|
|
f.Execute(ctx, `{}`) // runs; crashes at charge
|
|
pending, _ := f.Pending(ctx) // the run, checkpointed at "charge"
|
|
f.Resume(ctx, pending[0].ID) // continues from charge to the end
|
|
```
|
|
|
|
- **`State`** carries a typed payload (`Set`/`Scan`) plus a `Stage`
|
|
marker — the resume point.
|
|
- **`Checkpoint`** persists each `Run`. The built-in is store-backed and
|
|
keeps each flow's runs in their own store table (database `flow`, table
|
|
`checkout`) via `store.Scope`, so one flow's runs don't share a table
|
|
with another's — or with agent or service state. Point the default
|
|
store at Postgres or NATS KV and a run survives a real process restart,
|
|
or implement the interface to plug in Temporal, Restate, etc.
|
|
- A real step would be `flow.Call(service, endpoint)` (an RPC),
|
|
`flow.Dispatch(agent)` (hand off to an agent), or `flow.LLM(prompt)`
|
|
(one model turn). Here they're plain funcs so durability is the only
|
|
thing on display.
|
|
|
|
## Run
|
|
|
|
```bash
|
|
go run main.go
|
|
```
|
|
|
|
No LLM key required.
|