maestro

README

maestro

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).

Architecture

• Soloist (pkg/maestro/soloist) — single-replica writer. Owns the BlobStore (writes). Drives 3PC rounds. Tracks a player roster from heartbeats. State is in-memory only; restart loses currentVersion and the cluster has no committed version until the next Publish.
• Player (pkg/maestro/player) — N replicas. Heartbeat to soloist. Subscribe to round broadcasts. Invoke a user-supplied StageHandler on stage / activate / abort. Download manifest files via BlobReader.
• Transport (pkg/maestro/transport) — typed pub/sub bundle over goflux+NATS. Eight goflux.Topic[T] fields; NewTransport(*nats.Conn) wires them.
• BlobStore / BlobReader (pkg/maestro/blobstore) — pluggable. localfs ships in-box: the soloist serves files over an internal HTTP handler; players fetch via localfs.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.

The 3PC protocol

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.

Implementation guide

1. Implement StageHandler

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 Stage votes against the round.
• FileSource.Open returns a stream whose bytes are sha256-verified against the manifest as you read. Do not re-hash.
• Abort may fire for a version that never staged successfully — treat as a no-op.

2. Wire a Player into a foomo/keel server

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.

3. Wire a Soloist into a foomo/keel server

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.

Configuration

player.Options

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.

soloist.Options

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.

Operational notes

• 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 the expected set 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 become Wired() but never Ready() until the next Publish. If cold-start data is required, have your soloist Publish on 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=1 Deployment with stable DNS. Restart = brief disconnect; players reconnect and resync if drifted.

How to Contribute

Contributions are welcome! Please read the contributing guide.

See CONTRIBUTING.md for details.

License

Distributed under MIT License, see LICENSE for details.

Made with ♥ foomo by bestbytes

Documentation

Overview

Package maestro replicates a single in-memory state from one writer (github.com/foomo/maestro/pkg/soloist.Soloist) to every replica (github.com/foomo/maestro/pkg/player.Player) using a three-phase commit. One soloist writes the score; every player turns the page together — each either flips to the new Version or keeps the old one, so no player ever serves a half-applied update.

Control-plane traffic (round coordination, votes, heartbeats) is small typed messages carried by github.com/foomo/maestro/pkg/transport. File bytes move separately through a pluggable github.com/foomo/maestro/pkg/blobstore.BlobStore; this package's own wire format never carries application data, only a Manifest describing named files by hash and size.

Application code implements github.com/foomo/maestro/pkg/player.StageHandler to decode a Manifest's files into whatever Go value it wants replicated, and calls github.com/foomo/maestro/pkg/soloist.Soloist.Publish to publish a new version from the writer side.

There is no leader election: the Soloist is a single, designated writer (typically a replicas=1 deployment). See the package documentation for github.com/foomo/maestro/pkg/soloist and github.com/foomo/maestro/pkg/player for the two roles, and https://foomo.github.io/maestro/ for the full guide.

Index

• Variables
• type Manifest
  • func (m Manifest) Validate() error
• type ManifestFile
• type Version
  • func (v Version) String() string

Variables

var (
	ErrNoPlayers         = errors.New("maestro: no players in roster")
	ErrAbort             = errors.New("maestro: round aborted")
	ErrGenStale          = errors.New("maestro: stale generation token")
	ErrManifestMismatch  = errors.New("maestro: manifest validation failed")
	ErrBlobstoreMismatch = errors.New("maestro: blobstore kind mismatch between soloist and player")
	ErrDuplicateInstance = errors.New("maestro: duplicate instance id in roster")
	ErrRoundInFlight     = errors.New("maestro: another round is in flight")
	ErrUnsafeName        = errors.New("maestro: manifest file name failed path safety check")
)

Types

type Manifest

type Manifest struct {
	Version   Version        `msgpack:"version"`
	Files     []ManifestFile `msgpack:"files"`
	TotalSize int64          `msgpack:"total_size"`
}

func (Manifest) Validate

func (m Manifest) Validate() error

type ManifestFile

type ManifestFile struct {
	Name string `msgpack:"name"`
	Hash string `msgpack:"hash"`
	Size int64  `msgpack:"size"`
}

type Version

type Version string

Version is an opaque, content-addressable identifier for a Manifest.

func (Version) String

func (v Version) String() string