One soloist writes the score. Every player turns the page together.
Atomic in-memory state replication for Go, from one writer to every replica.
Replicate a single in-memory state from one writer to many readers, atomically. The state is whatever Go
value your service holds in memory — a catalog, a routing table, a feature flag set. Maestro coordinates a three-phase
commit so every reader either flips to the new version together or keeps the old one; no reader ever serves a
half-applied update. Bytes flow through a pluggable
BlobStore (HTTP, S3, …); only control plane traffic touches the message bus (NATS today).
- Soloist (
pkg/maestro/soloist) — single-replica writer. Owns theBlobStore(writes). Drives 3PC rounds. Tracks a player roster from heartbeats. State is in-memory only; restart losescurrentVersionand the cluster has no committed version until the nextPublish. - Player (
pkg/maestro/player) — N replicas. Heartbeat to soloist. Subscribe to round broadcasts. Invoke a user-suppliedStageHandleron stage / activate / abort. Download manifest files viaBlobReader. - Transport (
pkg/maestro/transport) — typed pub/sub bundle over goflux+NATS. Eightgoflux.Topic[T]fields;NewTransport(*nats.Conn)wires them. - BlobStore / BlobReader (
pkg/maestro/blobstore) — pluggable.localfsships in-box: the soloist serves files over an internal HTTP handler; players fetch vialocalfs.NewClient(httpBase).
┌──────────┐ heartbeats (NATS) ┌─────────┐
│ Player │ ───────────────────▶ │ Soloist │
│ (× N) │ ◀─────────────────── │ (× 1) │
└────┬─────┘ 3PC (NATS) └────┬────┘
│ │
│ HTTP GET /blob/{v}/{name} │
▼ ▼
┌──────────────────────────────────────┐
│ BlobStore (localfs / s3 / …) │
└──────────────────────────────────────┘
No leader election. Soloist is replicas=1. Restart = brief NATS disconnect; players reconnect and resync if drifted.
sequenceDiagram
participant S as Soloist
participant B as BlobStore
participant P as Player(s)
Note over P, S: Heartbeat (every HeartbeatPeriod)
P ->> S: Heartbeat{currentVersion}
Note over S: Publish(files) →<br/>compute Version → write blobs
S ->> B: Write each file
S ->> B: Finalize(v, manifest)
Note over S, P: Phase 1 — CanCommit
S ->> P: CanCommit{rid, target=v, manifest}
P ->> P: Validate manifest, stash pending
P -->> S: Vote{ok=true}
Note over S, P: Phase 2 — PreCommit (download + stage)
S ->> P: PreCommit{rid, target=v}
P ->> B: GET each blob (parallel)
P ->> P: StageHandler.Stage(v, manifest, src)
P -->> S: Staged{ok=true}
Note over S, P: Phase 3 — DoCommit (each player activates, then confirms)
S ->> P: DoCommit{rid, target=v}
P ->> P: StageHandler.Activate(v)
P -->> S: Committed{ok=true}
Note over S: setCurrent(v)
Abort path. If any player votes ok=false in phase 1, or any Staged
reply fails in phase 2, the soloist publishes Abort{reason} for the round. Every player that has stashed a pending
artifact for the round calls
StageHandler.Abort(v) to discard it. Phase 3 failures (slow / missing
Committed replies) are not aborted — the soloist marks the player dirty for the next resync.
Expected set. A round targets the players that are alive and wired — heartbeating within HeartbeatWindow with
their round subscriptions acknowledged by the broker. A player still starting up advertises Heartbeat.NotWired and is
left out rather than aborting the round for everyone; it resyncs once ready. Among the players that can vote, every
phase is unanimous: one ok=false or one silence aborts the round.
Silent commit. If the roster is empty at Publish time, the soloist skips 3PC and writes currentVersion directly.
Late-joining players resync. If players are present but none is wired yet, Publish fails instead — committing
silently would advance past a version they never saw.
Resync. Every RosterScanTick, the soloist scans heartbeats. Any alive, wired player whose currentVersion differs
from Current() is targeted with a fresh 3PC round, debounced by ResyncDebounce.
The Player calls these three methods. Stage builds an in-memory artifact from the manifest's files; Activate
promotes it; Abort discards it.
type myHandler struct {
cur atomic.Pointer[Catalog]
pending map[maestro.Version]*Catalog
mu sync.Mutex
}
func (h *myHandler) Stage(ctx context.Context, v maestro.Version, m maestro.Manifest, src player.FileSource) error {
r, err := src.Open("catalog.json") // hash-verified by the framework
if err != nil {
return err
}
defer r.Close()
var c Catalog
if err := json.NewDecoder(r).Decode(&c); err != nil {
return err
}
h.mu.Lock()
h.pending[v] = &c
h.mu.Unlock()
return nil
}
func (h *myHandler) Activate(ctx context.Context, v maestro.Version) error {
h.mu.Lock()
defer h.mu.Unlock()
p, ok := h.pending[v]
delete(h.pending, v)
if !ok {
return fmt.Errorf("no pending for %q", v)
}
h.cur.Store(p)
return nil
}
func (h *myHandler) Abort(_ context.Context, v maestro.Version) error {
h.mu.Lock()
delete(h.pending, v)
h.mu.Unlock()
return nil
}
func (h *myHandler) Current() *Catalog { return h.cur.Load() }Notes:
- Returning an error from
Stagevotes against the round. FileSource.Openreturns bytes already sha256-verified against the manifest beforeStageruns. Do not re-hash.Abortmay fire for a version that never staged successfully — treat as a no-op.
svr := keel.NewServer(
keel.WithHTTPHealthzService(true),
keel.WithHTTPPrometheusService(true),
)
l := svr.Logger()
nc, err := keelnats.Connect(svr, natsURL)
log.Must(l, err)
h := &myHandler{pending: make(map[maestro.Version]*Catalog)}
pl, err := player.New(player.Options{
Logger: l,
Transport: transport.NewTransport(nc),
BlobReader: localfs.NewClient(soloistHTTPBase),
InstanceID: instanceID,
StageHandler: h,
})
log.Must(l, err)
svr.AddService(pl) // Start blocks for service lifetime
svr.AddCloser(pl) // Close on SIGTERM
wired := healthz.NewHealthzerFn(func(_ context.Context) error {
if !pl.Wired() {
return errors.New("player not wired")
}
return nil
})
svr.AddReadinessHealthzers(wired)
svr.AddLivenessHealthzers(wired)
svr.AddPublicHTTPService(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
c := h.Current()
if c == nil {
http.Error(w, "no state", http.StatusServiceUnavailable)
return
}
_ = json.NewEncoder(w).Encode(c)
}))
svr.Run()Gate both probes on Wired() — true once the broker has acknowledged every round subscription, i.e. the player will
receive the next round. Not Ready() — Ready() flips only after the first DoCommit lands, so a cold cluster
(soloist restarted, no Publish yet) leaves the pod permanently NotReady. Surface "no data yet" through the public
HTTP handler instead.
svr := keel.NewServer(
keel.WithInitService(keelnatsservice.MustNewEmbeddedServer()),
keel.WithHTTPHealthzService(true),
)
l := svr.Logger()
nc, err := keelnats.Connect(svr, keelnatsservice.DefaultEmbeddedServerURL)
log.Must(l, err)
bs, err := localfs.NewStore(localfs.Config{DataDir: "/var/lib/maestro/data"})
log.Must(l, err)
svr.AddInternalHTTPService(bs.Handler()) // players GET blobs here
sol, err := soloist.New(soloist.Options{
Logger: l,
BlobStore: bs,
Transport: transport.NewTransport(nc),
InstanceID: instanceID,
})
log.Must(l, err)
svr.AddService(sol)
svr.AddCloser(sol)
ready := healthz.NewHealthzerFn(func(_ context.Context) error {
if !sol.Ready() {
return errors.New("soloist not ready")
}
return nil
})
svr.AddReadinessHealthzers(ready)
svr.AddLivenessHealthzers(ready)
// publish handler: POST → sol.Publish(ctx, []soloist.File{...})
svr.Run()The embedded NATS server lives inside the soloist pod. Players reach it over the cluster network.
localfs.NewStore(...).Handler() is the http.Handler
that serves blobs to player downloaders.
| Field | Default | Notes |
|---|---|---|
HeartbeatPeriod |
5s |
Lower = faster roster updates, more NATS traffic. |
DownloadConcurrency |
4 |
Parallel blob fetches inside one PreCommit. |
InstanceID |
— | Unique per pod. Use hostname. |
| Field | Default | Notes |
|---|---|---|
HeartbeatWindow |
15s |
Player is "alive" if a heartbeat arrived within this window. |
RosterScanTick |
5s |
How often the resync loop wakes. |
ResyncDebounce |
10s |
Minimum gap between two resync rounds. |
CanCommitTimeout |
10s |
Phase 1 deadline. |
StageTimeout |
~2× size / 10 MiB/s, ≥60s, ≤30m |
Phase 2 deadline. Adaptive to manifest total size. |
DoCommitTimeout |
10s |
Phase 3 deadline. Stragglers are marked dirty, not aborted. |
All three of Player, Soloist, and Transport accept a *zap.Logger, OTel
metric.MeterProvider, and trace.TracerProvider via their Options.
- Heartbeat-before-subscribe. A fresh player publishes its first heartbeat before its subscribers finish wiring, so
for a moment it is in the roster but cannot answer a round. It advertises this (
Heartbeat.NotWired) and the soloist leaves it out of theexpectedset until its subscriptions are live, then resyncs it. Without that gate every rolling replica would abort publishes for every other player while it started. Nothing to do. - Soloist restart loses
currentVersion. New players becomeWired()but neverReady()until the nextPublish. If cold-start data is required, have your soloistPublishon boot from a persistent source. - No rollback. Forward-only. A bad publish is fixed by the next publish.
- No leader election. Soloist is a
replicas=1Deployment with stable DNS. Restart = brief disconnect; players reconnect and resync if drifted.
Contributions are welcome! Please read the contributing guide.
See CONTRIBUTING.md for details.
Distributed under MIT License, see LICENSE for details.