amadan

eleven / messenger Public

server: the activator mesh — multi-box placement for hibernating instances (#550 Part 4 v2, Phase 3) (#678)

Multiple router boxes now cooperate so a hibernating instance wakes on
exactly one box, a request landing on the wrong box reverse-proxies to the
owner, and a dead box's instances wake elsewhere. Four mechanisms:

- Membership: S3 heartbeat objects (internal/activator/heartbeat.go) under
  <prefix>/mesh/members — the bucket is already the shared substrate, so no
  new ports/protocols between boxes. The key IS the record (base64url of the
  peer URL; one LIST, no GETs), LastModified is the liveness clock, stale
  after 3 missed beats. Identity = LCHAT_MESH_SELF (a peer-reachable base
  URL); unset = single box, zero mesh codepaths (tripwire-tested).
- Rendezvous (HRW) hashing: owner(host) is a pure function of the live
  member set; a membership change moves only ~1/N of hosts.
- The lease stays the truth; hashing is only a hint. A wrong-believer loses
  the CAS race loudly and 503s — membership disagreement can't split-brain.
- Proxy-to-owner: Host preserved (the peer routes by Host); https peers get
  SNI + cert verification against the *instance* hostname with TCP pinned to
  the peer's address (never InsecureSkipVerify); a forwarded marker caps
  proxying at one hop. Dead/stale owner falls through to a local wake.

Drain (activatorSvc.shutdown) deletes the heartbeat FIRST, then checkpoints
and releases as before — deploys, box failure, and scale-down are one code
path. Due-time sweeps only fire on the HRW owner. Also hardens the final
checkpoint: LeasedReplicator.Close now forces a SyncAndWait so a drain right
after a wake can't ship an empty replica (litestream only final-syncs an
initialized DB), and gains Abandon() so tests can simulate hard box death.

Tested: HRW determinism/redistribution, staleness, routing decision, proxy
rewrite semantics, drain ordering (order-pinned), due-sweep gating, mesh-off
tripwire — plus two live two-box kill-a-box tests against real S3 (gated on
LCHAT_TEST_S3_BUCKET): drain failover (survivor takes over in ~2s) and hard
kill (heartbeat rots, lease expires, survivor restores with data intact in
~30s).
Co-authored-by: Claude Fable 5 <noreply@anthropic.com>
Paul Campbell 3542a7ea452e2c68a14b79da8fcc131c5603803a parent 70f8bf3
8 files changed, +1627 −3
  • activator.go +67 −0
  • internal/activator/heartbeat.go +137 −0
  • internal/activator/heartbeat_test.go +23 −0
  • internal/activator/replica.go +22 −0
  • mesh.go +356 −0
  • mesh_live_test.go +365 −0
  • mesh_test.go +633 −0
  • router.go +24 −3
diff --git a/activator.go b/activator.go
index 2072cdf..7c839bd 100644
--- a/activator.go
+++ b/activator.go
@@ -109,6 +109,14 @@ func (c *hibernateConfig) hibernatePrefix(host string) string {
return path.Join(c.Prefix, host)
}
+// meshPrefix is where the mesh's membership heartbeats live, beside the
+// per-instance prefixes — the bucket is the mesh's only shared substrate.
+// It can't collide with hibernatePrefix: instance prefixes are public FQDNs
+// (they contain a dot), and the bare label "mesh" never is one.
+func (c *hibernateConfig) meshPrefix() string {
+ return path.Join(c.Prefix, "mesh/members")
+}
+
// --- backend seam ------------------------------------------------------------
//
// The state machine's side effects (S3 lease, restore, process spawn) sit
@@ -261,6 +269,13 @@ type activatorSvc struct {
due actDueStore
now func() time.Time // a seam for the unit tests' clock
+ // mesh, when non-nil, is the multi-box placement layer (mesh.go): requests
+ // for hosts whose HRW owner is another live box are proxied there, and the
+ // due-time sweeper only wakes hosts this box owns. Nil (LCHAT_MESH_SELF
+ // unset) is a single box — hashing would pick self everywhere, so no mesh
+ // codepath is touched at all.
+ mesh *meshSvc
+
mu sync.Mutex
entries map[string]*actEntry
@@ -421,6 +436,17 @@ func (a *activatorSvc) beginSidecarRequest(ctx context.Context, rt Route) error
}
}
+// hasLiveEntry reports whether host has an entry in any state — the mesh's
+// "already living here" check (waking and stopping both count: a waking host
+// is committed to this box, and a stopping one still holds the lease, so
+// proxying it away would only trade a short wait for a loud lease race).
+func (a *activatorSvc) hasLiveEntry(host string) bool {
+ a.mu.Lock()
+ defer a.mu.Unlock()
+ _, ok := a.entries[host]
+ return ok
+}
+
func (a *activatorSvc) releaseFunc(e *actEntry) func() {
var once sync.Once
return func() {
@@ -721,6 +747,19 @@ func (a *activatorSvc) sweepOnce() {
if _, live := a.entries[rt.Host]; live {
continue
}
+ // Mesh: only the HRW owner due-wakes a host — otherwise every box
+ // sweeping a due entry would race a wake per tick (the lease would let
+ // exactly one win, but loudly, forever). If the owner box is dead its
+ // heartbeat goes stale and the next tick's membership makes someone
+ // else the owner — eventual, bounded by staleAfter, and a scheduled
+ // delivery is late by at most that. (Today each box records next_due
+ // in its own registry, so the recorder and the owner are usually the
+ // same box; a host whose ownership moved while asleep waits for the
+ // registry-sync work — or a member request — to fire. Known gap,
+ // deliberately deferred with the shared-registry question.)
+ if a.mesh != nil && !a.mesh.owns(rt.Host) {
+ continue
+ }
log.Printf("activator: %s has a scheduled delivery due — waking it", rt.Host)
e := &actEntry{host: rt.Host, route: rt, state: actWaking, ready: make(chan struct{})}
a.entries[rt.Host] = e
@@ -781,6 +820,14 @@ func (a *activatorSvc) retireExcept(keep map[string]bool) {
// the leases until TTL expiry and leave the sockets bound against the next
// incarnation's spawns). Blocks until all stops complete or ctx expires.
func (a *activatorSvc) shutdown(ctx context.Context) {
+ // Drain step 1, order-pinned: leave the mesh membership FIRST — the
+ // heartbeat is deleted before any checkpoint or lease release below, so
+ // peers stop proxying here immediately and their wakes succeed the moment
+ // each lease releases. Deploys, box failure drills, and scale-down all
+ // ride this same path.
+ if a.mesh != nil {
+ a.mesh.stop(ctx)
+ }
if a.sweepStop != nil {
close(a.sweepStop)
<-a.sweepDone
@@ -839,6 +886,26 @@ func (a *activatorSvc) serveHTTP(w http.ResponseWriter, r *http.Request, rt Rout
return
}
+ // Mesh placement (mesh.go): a member request for a host with no live local
+ // entry and an HRW owner on another live box is proxied there instead of
+ // woken here. A forwarded request is never re-proxied (one hop max — see
+ // meshForwardedHeader). A host with ANY live local entry stays local: an
+ // awake instance finishes its natural life where it is, and placement
+ // converges at the next sleep boundary — its state is already in S3, so
+ // "migrating" is just waking on the new owner. The check-then-proxy is
+ // deliberately not atomic with the entry map: a race sends one request the
+ // long way, or wakes a not-the-hint box, and the lease arbitrates exactly
+ // as it would for real membership disagreement. (Sidecar-marked traffic
+ // above never routes through the mesh — a co-spawned sidecar's instance is
+ // by construction local, and a stray marked request for a host living
+ // elsewhere is exactly the reconnect-racing-a-stop case the 503 exists for.)
+ if a.mesh != nil && r.Header.Get(meshForwardedHeader) == "" && !a.hasLiveEntry(rt.Host) {
+ if peer, remote := a.mesh.remoteOwner(rt.Host); remote {
+ a.mesh.proxyTo(w, r, peer)
+ return
+ }
+ }
+
release, err := a.beginRequest(r.Context(), rt)
if err != nil {
var held *errLeaseHeld
diff --git a/internal/activator/heartbeat.go b/internal/activator/heartbeat.go
new file mode 100644
index 0000000..e433d9c
--- /dev/null
+++ b/internal/activator/heartbeat.go
@@ -0,0 +1,137 @@
+// 🤖 Mesh membership heartbeats (issue #550 Part 4 v2, Phase 3 — "the
+// activator mesh"). Boxes learn who's alive from the bucket itself: each live
+// box PUTs one object under a shared members prefix on an interval, LISTs the
+// prefix to see its peers, and DELETEs its object on drain. S3 was chosen over
+// gossip (memberlist) deliberately: the bucket is already the mesh's shared
+// substrate — the leases and replicas live there — so membership needs no new
+// ports or protocols between boxes, which may not even share a network (the
+// current prod box plus a future AWS pool). Membership changes are rare
+// events; a ~15s poll is plenty.
+//
+// The KEY is the record: an object key is base64url(peer URL), so one LIST —
+// no per-member GETs — yields the whole membership, and LastModified (the
+// bucket's clock, not the writer's) is the liveness timestamp, so writer
+// clock skew never enters the staleness judgement. The body duplicates the
+// URL as JSON purely for operators poking at the bucket. On a versioned
+// bucket the beats churn noncurrent versions; the standard keep-N lifecycle
+// prunes them.
+
+package activator
+
+import (
+ "bytes"
+ "context"
+ "encoding/base64"
+ "encoding/json"
+ "fmt"
+ "path"
+ "strings"
+ "time"
+
+ "github.com/aws/aws-sdk-go-v2/aws"
+ awsconfig "github.com/aws/aws-sdk-go-v2/config"
+ awss3 "github.com/aws/aws-sdk-go-v2/service/s3"
+)
+
+// Heartbeat is one box's liveness record in the mesh membership prefix.
+type Heartbeat struct {
+ Peer string // the box's peer base URL — its mesh identity
+ LastSeen time.Time // the object's S3 LastModified — the bucket's clock
+}
+
+// HeartbeatStore reads and writes box heartbeats under s3://bucket/prefix/.
+type HeartbeatStore struct {
+ client *awss3.Client
+ bucket string
+ prefix string
+}
+
+// NewHeartbeatStore builds a store on the default AWS credential chain, like
+// AcquireLease. Region may be empty; the chain resolves it.
+func NewHeartbeatStore(ctx context.Context, bucket, prefix, region string) (*HeartbeatStore, error) {
+ var opts []func(*awsconfig.LoadOptions) error
+ if region != "" {
+ opts = append(opts, awsconfig.WithRegion(region))
+ }
+ cfg, err := awsconfig.LoadDefaultConfig(ctx, opts...)
+ if err != nil {
+ return nil, fmt.Errorf("load aws config: %w", err)
+ }
+ return &HeartbeatStore{
+ client: awss3.NewFromConfig(cfg),
+ bucket: bucket,
+ prefix: strings.Trim(prefix, "/"),
+ }, nil
+}
+
+// encodeHeartbeatKey / decodeHeartbeatKey turn a peer URL into an S3-safe key
+// leaf and back (URLs carry '/' and ':', which would fake folder structure).
+func encodeHeartbeatKey(peer string) string {
+ return base64.RawURLEncoding.EncodeToString([]byte(peer))
+}
+
+func decodeHeartbeatKey(leaf string) (string, error) {
+ b, err := base64.RawURLEncoding.DecodeString(leaf)
+ if err != nil {
+ return "", err
+ }
+ return string(b), nil
+}
+
+func (s *HeartbeatStore) key(peer string) string {
+ return path.Join(s.prefix, encodeHeartbeatKey(peer))
+}
+
+// Put writes (or refreshes) this box's heartbeat. The write itself is the
+// timestamp — LastModified advances on every beat.
+func (s *HeartbeatStore) Put(ctx context.Context, peer string) error {
+ body, err := json.Marshal(map[string]string{"peer": peer})
+ if err != nil {
+ return err
+ }
+ _, err = s.client.PutObject(ctx, &awss3.PutObjectInput{
+ Bucket: aws.String(s.bucket),
+ Key: aws.String(s.key(peer)),
+ Body: bytes.NewReader(body),
+ })
+ return err
+}
+
+// List returns every heartbeat under the prefix. Objects whose key leaf isn't
+// a decodable peer record are skipped — a foreign object in the prefix must
+// never poison membership.
+func (s *HeartbeatStore) List(ctx context.Context) ([]Heartbeat, error) {
+ var out []Heartbeat
+ p := awss3.NewListObjectsV2Paginator(s.client, &awss3.ListObjectsV2Input{
+ Bucket: aws.String(s.bucket),
+ Prefix: aws.String(s.prefix + "/"),
+ })
+ for p.HasMorePages() {
+ page, err := p.NextPage(ctx)
+ if err != nil {
+ return nil, err
+ }
+ for _, obj := range page.Contents {
+ peer, err := decodeHeartbeatKey(path.Base(aws.ToString(obj.Key)))
+ if err != nil {
+ continue
+ }
+ var last time.Time
+ if obj.LastModified != nil {
+ last = *obj.LastModified
+ }
+ out = append(out, Heartbeat{Peer: peer, LastSeen: last})
+ }
+ }
+ return out, nil
+}
+
+// Delete removes this box's heartbeat — the FIRST step of a drain, before any
+// checkpoint or lease release, so peers take over immediately.
+func (s *HeartbeatStore) Delete(ctx context.Context, peer string) error {
+ _, err := s.client.DeleteObject(ctx, &awss3.DeleteObjectInput{
+ Bucket: aws.String(s.bucket),
+ Key: aws.String(s.key(peer)),
+ })
+ return err
+}
diff --git a/internal/activator/heartbeat_test.go b/internal/activator/heartbeat_test.go
new file mode 100644
index 0000000..25ff927
--- /dev/null
+++ b/internal/activator/heartbeat_test.go
@@ -0,0 +1,23 @@
+package activator
+
+import "testing"
+
+// The key IS the membership record (one LIST, no GETs), so the encoding must
+// round-trip any peer URL and reject foreign objects.
+func TestHeartbeatKeyRoundTrip(t *testing.T) {
+ for _, peer := range []string{
+ "http://10.0.0.5:8080",
+ "https://box-a.internal",
+ "https://box-b.mesh.example.com:8443",
+ } {
+ leaf := encodeHeartbeatKey(peer)
+ if got, err := decodeHeartbeatKey(leaf); err != nil || got != peer {
+ t.Fatalf("round-trip %q: got %q, %v", peer, got, err)
+ }
+ }
+ // A foreign object's key leaf must fail decode (List skips it) rather than
+ // turn into a phantom member.
+ if _, err := decodeHeartbeatKey("not/base64!"); err == nil {
+ t.Fatal("garbage key leaf decoded without error")
+ }
+}
diff --git a/internal/activator/replica.go b/internal/activator/replica.go
index aba972f..177fbe0 100644
--- a/internal/activator/replica.go
+++ b/internal/activator/replica.go
@@ -212,9 +212,31 @@ func (r *LeasedReplicator) SyncAndWait(ctx context.Context) error {
// Err reports whether lease renewal has failed (leadership lost).
func (r *LeasedReplicator) Err() error { return r.rn.Err() }
+// Abandon stops lease renewal and replication WITHOUT the final sync or the
+// lease release — the wreckage a crashed box leaves behind: lock.json simply
+// expires on its TTL, and the replica holds whatever was last synced. It
+// exists so tests can simulate hard box death in-process; production code has
+// no reason to call it (a real crash needs no help).
+func (r *LeasedReplicator) Abandon(ctx context.Context) {
+ r.rn.stop() // deliberately no ReleaseLease — the TTL is the point
+ // litestream's only teardown is Close, which includes a sync a real crash
+ // would skip; it ships nothing as long as the caller killed the writer
+ // first, and reclaims the replication goroutines either way.
+ _ = r.store.Close(ctx)
+}
+
// Close stops replication (final sync included via litestream's DB.Close)
// and releases the lease so the next claimant can acquire immediately.
func (r *LeasedReplicator) Close(ctx context.Context) error {
+ // Force one full sync before teardown. litestream's DB.Close only
+ // final-syncs an *initialized* DB, and initialization happens on the
+ // monitor's first pass — so a checkpoint within ~a second of Replicate
+ // could silently ship nothing (a drain right after a wake, caught by the
+ // mesh live test: the replica held no restorable snapshot). SyncAndWait
+ // forces init + a replica sync; its error is deliberately dropped — with
+ // the DB now initialized, store.Close's own final sync is the one that
+ // counts, and its error is returned.
+ _ = r.db.SyncAndWait(ctx)
err := r.store.Close(ctx)
lease := r.rn.stop()
diff --git a/mesh.go b/mesh.go
new file mode 100644
index 0000000..1e686a6
--- /dev/null
+++ b/mesh.go
@@ -0,0 +1,356 @@
+package main
+
+// The activator mesh (issue #550 Part 4 v2, Phase 3): multiple router boxes
+// cooperate so a hibernating instance wakes on exactly one box, a request
+// landing on the wrong box is reverse-proxied to the owner, and a dead box's
+// instances wake elsewhere. Four mechanisms, deliberately this simple:
+//
+// 1. Membership — boxes learn who's alive from S3 heartbeat objects
+// (internal/activator/heartbeat.go), not gossip: the bucket is already the
+// shared substrate, so no new ports or protocols between boxes (which may
+// not even share a network). Each box PUTs mesh/members/<box> every
+// meshHeartbeatInterval and LISTs the prefix; a record older than
+// meshStaleAfter is a dead box. A box's identity is LCHAT_MESH_SELF — a
+// base URL its peers can reach it at; the env's presence turns the mesh
+// on, its absence keeps today's single-box behavior byte-identical.
+// 2. Rendezvous (HRW) hashing — owner(host) is a pure function of the live
+// member set (hrwOwner): no routing table, no directory service, and a
+// membership change moves only ~1/N of hosts (only onto a joiner, only
+// off a leaver).
+// 3. The lease stays the truth; hashing is only a hint. The owner claims the
+// per-host S3 lease exactly as on one box; a box that wrongly believes
+// it's the owner loses the CAS race loudly (503 + Retry-After) and the
+// client's retry lands right. Transient membership disagreement therefore
+// cannot split-brain — single-writer was never the mesh's job.
+// 4. Proxy-to-owner — a request for a sleeping host whose HRW owner is
+// another live box is reverse-proxied to that box's peer URL, instance
+// hostname preserved (the peer routes by Host like any front door). A
+// dead/stale owner drops out of the member set, so the hash reassigns and
+// the request falls through to a local wake — failover IS the routine
+// wake path.
+//
+// Drain (activatorSvc.shutdown) deletes the heartbeat FIRST, then checkpoints
+// and releases as today — peers take over immediately, and deploys, box
+// failure, and scale-down are the same code path.
+
+import (
+ "context"
+ "errors"
+ "fmt"
+ "hash/fnv"
+ "io"
+ "log"
+ "net"
+ "net/http"
+ "net/http/httputil"
+ "net/url"
+ "os"
+ "strings"
+ "sync"
+ "time"
+
+ "lchat/internal/activator"
+)
+
+// meshForwardedHeader marks a request one box already proxied to another:
+// "serve where you stand, never re-proxy". One hop max — if two boxes
+// transiently disagree about ownership, the worst case is a wrong-box wake
+// the lease arbitrates, never a proxy loop. An outsider spoofing it merely
+// pins its own request to the box it hit, which is exactly what every request
+// could do before the mesh existed; the lease still guarantees one writer.
+// Stripped before any backend sees it (unixProxy), like the sidecar marker.
+const meshForwardedHeader = "X-Lchat-Mesh-Forwarded"
+
+const (
+ // meshHeartbeatInterval paces the PUT + LIST beat. Membership changes are
+ // rare events; this is liveness, not coordination.
+ meshHeartbeatInterval = 15 * time.Second
+ // meshStaleAfter is how old a heartbeat may be before its box is presumed
+ // dead — three missed beats. It's judged against the reader's clock but
+ // the timestamp is S3's LastModified (the bucket's clock), so only
+ // reader-vs-S3 skew matters, tiny against 45s. Together with the ~30s
+ // lease TTL this bounds failover: a hard-killed box's hosts wake elsewhere
+ // once its heartbeat is stale AND its leases expire.
+ meshStaleAfter = 3 * meshHeartbeatInterval
+)
+
+// meshStore is the S3 heartbeat surface (activator.HeartbeatStore in
+// production), an interface so the mesh's unit tests share an in-memory
+// "bucket" between boxes.
+type meshStore interface {
+ Put(ctx context.Context, peer string) error
+ List(ctx context.Context) ([]activator.Heartbeat, error)
+ Delete(ctx context.Context, peer string) error
+}
+
+type meshSvc struct {
+ self string // this box's peer base URL — its mesh identity
+ store meshStore
+ interval time.Duration
+ staleAfter time.Duration
+ now func() time.Time // a seam for the unit tests' clock
+
+ mu sync.Mutex
+ members []activator.Heartbeat // last LIST; staleness applied at read time
+ proxies map[string]*httputil.ReverseProxy
+ started bool
+
+ stopCh chan struct{}
+ done chan struct{}
+}
+
+func newMesh(self string, store meshStore) *meshSvc {
+ return &meshSvc{
+ self: self,
+ store: store,
+ interval: meshHeartbeatInterval,
+ staleAfter: meshStaleAfter,
+ now: time.Now,
+ proxies: make(map[string]*httputil.ReverseProxy),
+ stopCh: make(chan struct{}),
+ done: make(chan struct{}),
+ }
+}
+
+// meshFromEnv reads the box's mesh identity. LCHAT_MESH_SELF is a base URL
+// peer boxes can reach this box at, scheme included — http on a private
+// network, or https where the box presents a verifiable cert (see
+// newPeerProxy). Presence turns the mesh on; absent = single box, zero mesh
+// codepaths. The mesh rides on the hibernation bucket, so it requires
+// LCHAT_HIBERNATE_BUCKET.
+func meshFromEnv(hc *hibernateConfig) (*meshSvc, error) {
+ self := strings.TrimSpace(os.Getenv("LCHAT_MESH_SELF"))
+ if self == "" {
+ return nil, nil
+ }
+ if hc == nil {
+ return nil, errors.New("LCHAT_MESH_SELF set but LCHAT_HIBERNATE_BUCKET missing — the mesh rides on the hibernation bucket and only places hibernating instances")
+ }
+ self = strings.TrimRight(self, "/")
+ u, err := url.Parse(self)
+ if err != nil || u.Host == "" || (u.Scheme != "http" && u.Scheme != "https") || u.Path != "" {
+ return nil, fmt.Errorf("LCHAT_MESH_SELF %q: want a base URL like https://box-a.internal or http://10.0.0.5:8080", self)
+ }
+ store, err := activator.NewHeartbeatStore(context.Background(), hc.Bucket, hc.meshPrefix(), hc.Region)
+ if err != nil {
+ return nil, fmt.Errorf("mesh heartbeat store: %w", err)
+ }
+ return newMesh(self, store), nil
+}
+
+// start registers this box (one synchronous beat, so the first sweep and the
+// first request already see a membership view) and begins the heartbeat loop.
+func (m *meshSvc) start() {
+ m.beat()
+ m.mu.Lock()
+ m.started = true
+ m.mu.Unlock()
+ go func() {
+ defer close(m.done)
+ t := time.NewTicker(m.interval)
+ defer t.Stop()
+ for {
+ select {
+ case <-m.stopCh:
+ return
+ case <-t.C:
+ m.beat()
+ }
+ }
+ }()
+}
+
+// beat PUTs our heartbeat and refreshes the member view. Both failures are
+// soft by design: a missed PUT makes peers see us stale (they take over; the
+// lease arbitrates), a failed LIST keeps the last known view (a stale view
+// misroutes at worst — same arbitration).
+func (m *meshSvc) beat() {
+ timeout := m.interval
+ if timeout < 5*time.Second {
+ timeout = 5 * time.Second
+ }
+ ctx, cancel := context.WithTimeout(context.Background(), timeout)
+ defer cancel()
+ if err := m.store.Put(ctx, m.self); err != nil {
+ log.Printf("mesh: heartbeat put failed: %v — peers will see this box stale after %s", err, m.staleAfter)
+ }
+ members, err := m.store.List(ctx)
+ if err != nil {
+ log.Printf("mesh: membership list failed: %v — keeping the last known view", err)
+ return
+ }
+ m.mu.Lock()
+ m.members = members
+ m.mu.Unlock()
+}
+
+// halt stops the heartbeat loop without touching the heartbeat object — the
+// shape a crashed box leaves behind (its record rots until staleAfter). The
+// drain path is stop; tests use halt to simulate hard death.
+func (m *meshSvc) halt() {
+ m.mu.Lock()
+ started := m.started
+ m.started = false
+ m.mu.Unlock()
+ if !started {
+ return
+ }
+ close(m.stopCh)
+ <-m.done
+}
+
+// stop is the drain path: halt the loop (a beat racing the delete would
+// resurrect us), then delete our heartbeat so peers take over immediately.
+// The caller (activatorSvc.shutdown) checkpoints and releases AFTER this —
+// that ordering is the drain contract.
+func (m *meshSvc) stop(ctx context.Context) {
+ m.halt()
+ if err := m.store.Delete(ctx, m.self); err != nil {
+ log.Printf("mesh: heartbeat delete failed: %v — peers take over when it goes stale (%s)", err, m.staleAfter)
+ }
+}
+
+// liveBoxes is the current membership: every fresh heartbeat with a sane peer
+// URL, self always included — our own PUT may not have landed or listed yet,
+// and a box always counts itself, so the set is never empty and hrwOwner is
+// always defined. Staleness is applied at read time, so a box that dies
+// between polls drops out the moment its record ages past staleAfter.
+func (m *meshSvc) liveBoxes() []string {
+ now := m.now()
+ m.mu.Lock()
+ members := m.members
+ m.mu.Unlock()
+ boxes := []string{m.self}
+ for _, hb := range members {
+ if hb.Peer == m.self || now.Sub(hb.LastSeen) > m.staleAfter {
+ continue
+ }
+ // A garbage record (someone else's object decoding to nonsense) must
+ // not enter the hash — it would attract ~1/N of hosts and blackhole
+ // them behind an unproxyable "peer".
+ if u, err := url.Parse(hb.Peer); err != nil || u.Host == "" || (u.Scheme != "http" && u.Scheme != "https") {
+ continue
+ }
+ boxes = append(boxes, hb.Peer)
+ }
+ return boxes
+}
+
+// owns reports whether this box is host's HRW owner in the live member set —
+// the due-time sweeper's gate: only the owner due-wakes a host.
+func (m *meshSvc) owns(host string) bool {
+ return hrwOwner(host, m.liveBoxes()) == m.self
+}
+
+// remoteOwner returns the peer to proxy host to when its HRW owner is another
+// live box. ok=false means serve locally: we are the owner, or the previous
+// owner is dead/stale (it's out of the member set, so the hash has already
+// reassigned — possibly to us — and a local wake is the failover).
+func (m *meshSvc) remoteOwner(host string) (peer string, ok bool) {
+ if o := hrwOwner(host, m.liveBoxes()); o != m.self {
+ return o, true
+ }
+ return "", false
+}
+
+// proxyTo forwards one request to the owning peer, instance hostname
+// preserved. Proxies are built lazily, one per peer, and live for the mesh's
+// life — peer URLs are stable box identities.
+func (m *meshSvc) proxyTo(w http.ResponseWriter, r *http.Request, peer string) {
+ m.mu.Lock()
+ p := m.proxies[peer]
+ if p == nil {
+ u, err := url.Parse(peer)
+ if err != nil || u.Host == "" {
+ // liveBoxes filters garbage, so this is belt and braces.
+ m.mu.Unlock()
+ log.Printf("mesh: unusable peer URL %q for %s", peer, r.Host)
+ http.Error(w, "instance's owner box is unreachable; retry shortly", http.StatusServiceUnavailable)
+ return
+ }
+ p = newPeerProxy(m.self, u)
+ m.proxies[peer] = p
+ }
+ m.mu.Unlock()
+ p.ServeHTTP(w, r)
+}
+
+// newPeerProxy builds the reverse proxy to one peer box. The peer routes by
+// Host exactly like the public front door, so the instance hostname must
+// survive the hop; the scheme comes from the peer URL and is honored:
+//
+// - http peers (a private network between boxes): dial the peer address,
+// Host header = the instance host.
+// - https peers: SNI *and* certificate verification run against the
+// instance hostname — the peer presents its wildcard/per-host cert, which
+// proves it serves this host (it could never prove it IS 10.x.y.z). The
+// outgoing URL keeps the instance host and a custom dialer pins the TCP
+// connection to the peer's address — the same shape as the box's own
+// /etc/hosts loopback pin. InsecureSkipVerify is never used.
+func newPeerProxy(self string, peer *url.URL) *httputil.ReverseProxy {
+ dialAddr := peerDialAddr(peer)
+ return &httputil.ReverseProxy{
+ Rewrite: func(pr *httputil.ProxyRequest) {
+ pr.SetXForwarded()
+ pr.Out.URL.Scheme = peer.Scheme
+ if peer.Scheme == "https" {
+ pr.Out.URL.Host = pr.In.Host // SNI + cert verification target
+ } else {
+ pr.Out.URL.Host = peer.Host
+ }
+ pr.Out.Host = pr.In.Host // the peer routes by Host, like any client
+ pr.Out.Header.Set(meshForwardedHeader, self)
+ },
+ Transport: &http.Transport{
+ DialContext: func(ctx context.Context, network, _ string) (net.Conn, error) {
+ return (&net.Dialer{Timeout: 10 * time.Second}).DialContext(ctx, network, dialAddr)
+ },
+ // Bounded like unixProxy's pool, for the same leak reasons.
+ MaxIdleConns: 32,
+ IdleConnTimeout: 90 * time.Second,
+ },
+ ErrorHandler: func(w http.ResponseWriter, r *http.Request, err error) {
+ // An owner that died but hasn't gone stale yet (or a box-to-box
+ // network fault): 503 + retry. Within staleAfter the membership
+ // view drops it and retries wake the instance locally instead —
+ // staleness bounds this window.
+ log.Printf("mesh: proxy %s to %s: %v", r.Host, peer, err)
+ w.Header().Set("Retry-After", "5")
+ http.Error(w, "instance's owner box is unreachable; retry shortly", http.StatusServiceUnavailable)
+ },
+ }
+}
+
+// peerDialAddr is the TCP address a peer base URL dials to, default ports
+// applied — computed once per proxy, since the peer URL never changes.
+func peerDialAddr(peer *url.URL) string {
+ if peer.Port() != "" {
+ return peer.Host
+ }
+ if peer.Scheme == "https" {
+ return peer.Hostname() + ":443"
+ }
+ return peer.Hostname() + ":80"
+}
+
+// hrwOwner picks host's owner from boxes by rendezvous (highest-random-weight)
+// hashing: the box with the highest hash(box||host) wins. A pure function of
+// (host, member set) — every box computes the same answer from membership
+// alone, and when a box joins or leaves only ~1/N of hosts change owner (a
+// joiner only steals, a leaver's hosts scatter over the survivors; nothing
+// else moves). The result is a placement HINT, not a lock: the S3 lease stays
+// the single-writer truth, which is why this can be twenty lines.
+func hrwOwner(host string, boxes []string) string {
+ var best string
+ var bestScore uint64
+ for _, b := range boxes {
+ h := fnv.New64a()
+ io.WriteString(h, b)
+ h.Write([]byte{0}) // separator: ("ab","c") must not collide with ("a","bc")
+ io.WriteString(h, host)
+ if s := h.Sum64(); best == "" || s > bestScore || (s == bestScore && b > best) {
+ best, bestScore = b, s
+ }
+ }
+ return best
+}
diff --git a/mesh_live_test.go b/mesh_live_test.go
new file mode 100644
index 0000000..c3ade11
--- /dev/null
+++ b/mesh_live_test.go
@@ -0,0 +1,365 @@
+package main
+
+// 🤖 The live two-box mesh tests ("the kill-a-box demo", #550 Part 4 v2
+// Phase 3): TWO in-process boxes — two activator+router pairs on different
+// loopback ports, each spawning REAL instance processes — sharing one REAL S3
+// bucket for leases, replicas, and mesh heartbeats. They prove the mesh's
+// whole story:
+//
+// - a request landing on the NON-owner box is reverse-proxied to the HRW
+// owner, which wakes the instance and serves;
+// - killing the owner box moves its instance to the survivor with data
+// intact — in BOTH kill flavors:
+// drain (TestMeshLiveProxyAndDrainFailover): heartbeat deleted
+// first, checkpoint + lease release — the survivor takes
+// over on its next membership poll;
+// hard kill (TestMeshLiveHardKillFailover): the box just dies — its
+// heartbeat rots to staleness, its lease expires on TTL, and
+// the survivor's wake is gated by the lease until then.
+//
+// Gated on LCHAT_TEST_S3_BUCKET like the other live tests. Run with:
+//
+// eval "$(aws configure export-credentials --profile eleven-dev --format env)"
+// LCHAT_TEST_S3_BUCKET=<bucket> go test -run TestMeshLive -v .
+
+import (
+ "bytes"
+ "context"
+ "crypto/rand"
+ "encoding/hex"
+ "encoding/json"
+ "fmt"
+ "io"
+ "net/http"
+ "net/http/httptest"
+ "os"
+ "os/exec"
+ "path/filepath"
+ "strings"
+ "sync"
+ "testing"
+ "time"
+
+ "lchat/internal/activator"
+)
+
+// liveMeshBox is one box of the pair: its own activator (real s3Backend, real
+// child processes), registry, mesh, and front door.
+type liveMeshBox struct {
+ name string
+ self string // the front URL — this box's mesh identity
+ act *activatorSvc
+ mesh *meshSvc
+ front *httptest.Server
+ reg *Registry
+ route Route
+
+ shutdownOnce sync.Once
+}
+
+// shutdown drains the box exactly once — tests that dismantle a box by hand
+// (the hard kill) consume the Once so cleanup can't double-drain it.
+func (b *liveMeshBox) shutdown(ctx context.Context) {
+ b.shutdownOnce.Do(func() { b.act.shutdown(ctx) })
+}
+
+// setupMeshLive builds the shared world (S3 prefix, staged binary, env file)
+// and the two boxes. Both boxes register the same host, each pointing at its
+// OWN socket and DB path — a box must never see its peer's local files, so a
+// failover's restore provably comes from S3.
+func setupMeshLive(t *testing.T, host string) (boxA, boxB *liveMeshBox, adminToken string) {
+ t.Helper()
+ bucket := os.Getenv("LCHAT_TEST_S3_BUCKET")
+ if bucket == "" {
+ t.Skip("LCHAT_TEST_S3_BUCKET not set; skipping live two-box mesh test")
+ }
+ region := os.Getenv("LCHAT_TEST_S3_REGION")
+ if region == "" {
+ region = "eu-west-1"
+ }
+ rb := make([]byte, 4)
+ _, _ = rand.Read(rb)
+ prefix := fmt.Sprintf("mesh-live/%d-%s", time.Now().Unix(), hex.EncodeToString(rb))
+ t.Cleanup(func() { cleanupS3Prefix(t, bucket, prefix, region) })
+
+ // Sandbox the box-layout globals (shared by both in-process boxes: the
+ // staged binary, live symlink, and the instance env file are box-agnostic).
+ dataDir := t.TempDir()
+ sockDir := fmt.Sprintf("/tmp/lchat-mesh-%d", os.Getpid())
+ if err := os.MkdirAll(sockDir, 0o755); err != nil {
+ t.Fatal(err)
+ }
+ t.Cleanup(func() { os.RemoveAll(sockDir) })
+ optTmp := t.TempDir()
+ oldData, oldSock, oldOpt := instanceDataDir, instanceSocketDir, optDir
+ instanceDataDir, instanceSocketDir, optDir = dataDir, sockDir, optTmp
+ t.Cleanup(func() { instanceDataDir, instanceSocketDir, optDir = oldData, oldSock, oldOpt })
+
+ bin := filepath.Join(optTmp, "lchat-built")
+ build := exec.Command("go", "build", "-o", bin, ".")
+ build.Env = append(os.Environ(), "GOTOOLCHAIN=auto")
+ if out, err := build.CombinedOutput(); err != nil {
+ t.Fatalf("go build: %v\n%s", err, out)
+ }
+ if err := stageVersion(bin, "meshlive"); err != nil {
+ t.Fatalf("stage: %v", err)
+ }
+ if err := pointLive(host, "meshlive"); err != nil {
+ t.Fatalf("point live: %v", err)
+ }
+
+ adminToken = "meshlive-" + hex.EncodeToString(rb)
+ if err := writeEnvFile(filepath.Join(dataDir, host+".env"),
+ map[string]string{"LCHAT_ADMIN_TOKEN": adminToken}); err != nil {
+ t.Fatalf("env file: %v", err)
+ }
+
+ cfg := hibernateConfig{Bucket: bucket, Region: region, Prefix: prefix, Idle: 2 * time.Minute}
+ newBox := func(name string) *liveMeshBox {
+ box := &liveMeshBox{name: name}
+ var err error
+ box.reg, err = openRegistry(filepath.Join(t.TempDir(), "registry.db"))
+ if err != nil {
+ t.Fatalf("open registry: %v", err)
+ }
+ dbDir := filepath.Join(dataDir, name)
+ if err := os.MkdirAll(dbDir, 0o755); err != nil {
+ t.Fatal(err)
+ }
+ box.route = Route{Host: host, Kind: "instance",
+ Socket: filepath.Join(sockDir, name+".sock"),
+ DBPath: filepath.Join(dbDir, host+".db"),
+ Hibernate: true}
+ if err := box.reg.upsert(box.route); err != nil {
+ t.Fatalf("upsert: %v", err)
+ }
+ box.act = newActivator(cfg, &s3Backend{cfg: cfg}, box.reg)
+ rt, err := newRouter(box.reg, box.act)
+ if err != nil {
+ t.Fatalf("new router: %v", err)
+ }
+ box.front = httptest.NewServer(rt)
+ t.Cleanup(box.front.Close)
+ box.self = box.front.URL
+
+ hb, err := activator.NewHeartbeatStore(context.Background(), bucket, cfg.meshPrefix(), region)
+ if err != nil {
+ t.Fatalf("heartbeat store: %v", err)
+ }
+ box.mesh = newMesh(box.self, hb)
+ box.mesh.interval = time.Second
+ box.mesh.staleAfter = 3 * time.Second
+ box.act.mesh = box.mesh
+ box.mesh.start()
+ box.act.start()
+ t.Cleanup(func() {
+ ctx, cancel := context.WithTimeout(context.Background(), 60*time.Second)
+ defer cancel()
+ box.shutdown(ctx)
+ })
+ return box
+ }
+ boxA, boxB = newBox("box-a"), newBox("box-b")
+
+ // Both boxes must see the full membership before the test routes anything.
+ waitForLive(t, "both boxes to see each other", 15*time.Second, func() bool {
+ return len(boxA.mesh.liveBoxes()) == 2 && len(boxB.mesh.liveBoxes()) == 2
+ })
+ return boxA, boxB, adminToken
+}
+
+// pickOwner orders the pair as (owner, survivor) for host by the same HRW
+// hash the boxes themselves use.
+func pickOwner(host string, a, b *liveMeshBox) (owner, survivor *liveMeshBox) {
+ if hrwOwner(host, []string{a.self, b.self}) == b.self {
+ return b, a
+ }
+ return a, b
+}
+
+func meshGet(t *testing.T, front, host, path string) (int, string) {
+ t.Helper()
+ req, _ := http.NewRequest("GET", front+path, nil)
+ req.Host = host
+ resp, err := http.DefaultClient.Do(req)
+ if err != nil {
+ t.Fatalf("GET %s: %v", path, err)
+ }
+ defer resp.Body.Close()
+ body, _ := io.ReadAll(resp.Body)
+ return resp.StatusCode, string(body)
+}
+
+func meshSetGroupName(t *testing.T, front, host, token, name string) {
+ t.Helper()
+ body, _ := json.Marshal(map[string]string{"name": name})
+ req, _ := http.NewRequest("POST", front+"/api/admin/group", bytes.NewReader(body))
+ req.Host = host
+ req.Header.Set("Authorization", "Bearer "+token)
+ req.Header.Set("Content-Type", "application/json")
+ resp, err := http.DefaultClient.Do(req)
+ if err != nil {
+ t.Fatalf("set group name: %v", err)
+ }
+ defer resp.Body.Close()
+ if resp.StatusCode != 200 {
+ b, _ := io.ReadAll(resp.Body)
+ t.Fatalf("set group name: %d %s", resp.StatusCode, b)
+ }
+}
+
+// pollUntilServed hammers the survivor's front door until the instance
+// answers 200, logging every distinct (status, body) transition — the test's
+// narrative of the failover: proxy-to-dead-peer 503s, then lease-held 503s,
+// then the wake. Returns the winning body.
+func pollUntilServed(t *testing.T, front, host string, deadline time.Duration) string {
+ t.Helper()
+ start := time.Now()
+ last := ""
+ for time.Now().Before(start.Add(deadline)) {
+ code, body := meshGet(t, front, host, "/api/group")
+ if code == 200 {
+ t.Logf(" +%s: 200 — served", time.Since(start).Truncate(time.Millisecond))
+ return body
+ }
+ if cur := fmt.Sprintf("%d %s", code, strings.TrimSpace(body)); cur != last {
+ t.Logf(" +%s: %s", time.Since(start).Truncate(time.Millisecond), cur)
+ last = cur
+ }
+ time.Sleep(time.Second)
+ }
+ t.Fatalf("instance never served on the survivor within %s", deadline)
+ return ""
+}
+
+// The drain variant: proxy-to-owner, then an orderly box exit (heartbeat
+// deleted FIRST, checkpoint, lease release) and the survivor taking over on
+// its next membership poll — the same code path a deploy or scale-down runs.
+func TestMeshLiveProxyAndDrainFailover(t *testing.T) {
+ const host = "mesh-drain.test"
+ boxA, boxB, adminToken := setupMeshLive(t, host)
+ owner, survivor := pickOwner(host, boxA, boxB)
+ t.Logf("HRW owner of %s is %s (%s); survivor is %s (%s)", host, owner.name, owner.self, survivor.name, survivor.self)
+
+ // --- Act 1: a request to the NON-owner proxies to the owner and serves. ---
+ start := time.Now()
+ code, _ := meshGet(t, survivor.front.URL, host, "/api/group")
+ if code != 200 {
+ t.Fatalf("request via non-owner: %d, want 200", code)
+ }
+ t.Logf("request to the non-owner proxied + woke the instance on %s in %s", owner.name, time.Since(start).Truncate(time.Millisecond))
+ if owner.act.entryCount() != 1 || !owner.act.entryAwake(host) {
+ t.Fatalf("instance is not awake on the owner (entries=%d)", owner.act.entryCount())
+ }
+ if survivor.act.entryCount() != 0 {
+ t.Fatal("the non-owner woke a local copy instead of proxying")
+ }
+
+ // Write real data through the proxied path.
+ groupName := "Mesh failover " + hex.EncodeToString([]byte(host))
+ meshSetGroupName(t, survivor.front.URL, host, adminToken, groupName)
+ t.Logf("wrote group name %q through the non-owner's front door", groupName)
+
+ // --- Act 2: drain the owner. Front door dark first (runRouter shuts the
+ // HTTP servers down before the activator — a draining box accepts no new
+ // work), then the drain proper: heartbeat delete, checkpoint, release. ---
+ start = time.Now()
+ ctx, cancel := context.WithTimeout(context.Background(), 60*time.Second)
+ defer cancel()
+ owner.front.Close()
+ owner.shutdown(ctx)
+ t.Logf("owner %s drained in %s (heartbeat deleted, instance checkpointed, lease released)", owner.name, time.Since(start).Truncate(time.Millisecond))
+
+ // The heartbeat really is gone from the bucket — only the survivor beats.
+ hbs, err := survivor.mesh.store.List(context.Background())
+ if err != nil {
+ t.Fatalf("list heartbeats: %v", err)
+ }
+ for _, hb := range hbs {
+ if hb.Peer == owner.self {
+ t.Fatal("drained owner's heartbeat is still in the bucket")
+ }
+ }
+
+ // --- Act 3: the survivor takes over — wake local, restore from S3. ---
+ body := pollUntilServed(t, survivor.front.URL, host, 60*time.Second)
+ if !strings.Contains(body, groupName) {
+ t.Fatalf("survivor lost the data: %s does not contain %q", body, groupName)
+ }
+ if !survivor.act.entryAwake(host) {
+ t.Fatal("instance is not awake on the survivor")
+ }
+ t.Logf("survivor %s restored the instance from S3 with data intact: %s", survivor.name, body)
+}
+
+// The hard-kill variant: the owner box dies with no goodbye — its heartbeat
+// is left to rot (stale in staleAfter) and its lease to expire (TTL). The
+// survivor's requests walk the whole failure narrative: proxy to the corpse
+// (503), local wake gated by the still-held lease (503 + Retry-After), then
+// the lease expires and the instance wakes on the survivor with every byte
+// that had replicated.
+func TestMeshLiveHardKillFailover(t *testing.T) {
+ const host = "mesh-kill.test"
+ boxA, boxB, adminToken := setupMeshLive(t, host)
+ owner, survivor := pickOwner(host, boxA, boxB)
+ t.Logf("HRW owner of %s is %s (%s); survivor is %s (%s)", host, owner.name, owner.self, survivor.name, survivor.self)
+
+ // --- Act 1: wake on the owner and write data. ---
+ if code, _ := meshGet(t, owner.front.URL, host, "/api/group"); code != 200 {
+ t.Fatalf("wake on owner: %d, want 200", code)
+ }
+ groupName := "Hard kill survivor " + hex.EncodeToString([]byte(host))
+ meshSetGroupName(t, owner.front.URL, host, adminToken, groupName)
+
+ // Make the write durable in S3 before the "crash" — a real hard kill
+ // loses only the last seconds of unshipped WAL (that's the design's RPO);
+ // the test pins the data so the assertion is about failover, not timing.
+ owner.act.mu.Lock()
+ entry := owner.act.entries[host]
+ owner.act.mu.Unlock()
+ if entry == nil {
+ t.Fatal("no live entry on the owner")
+ }
+ rep := entry.rep.(*activator.LeasedReplicator)
+ syncCtx, syncCancel := context.WithTimeout(context.Background(), 30*time.Second)
+ if err := rep.SyncAndWait(syncCtx); err != nil {
+ syncCancel()
+ t.Fatalf("sync before kill: %v", err)
+ }
+ syncCancel()
+ t.Logf("wrote group name %q on the owner and confirmed it replicated to S3", groupName)
+
+ // --- Act 2: hard-kill the owner box. Everything at once, nothing orderly:
+ // front door gone, sweeper gone, heartbeat loop gone (object NOT deleted),
+ // instance child SIGKILLed, lease renewal abandoned (NOT released). ---
+ owner.front.Close()
+ close(owner.act.sweepStop)
+ <-owner.act.sweepDone
+ owner.mesh.halt() // the heartbeat object stays — it must rot to staleness
+ owner.act.mu.Lock()
+ delete(owner.act.entries, host)
+ owner.act.mu.Unlock()
+ proc := entry.proc.(*childProc)
+ if err := proc.cmd.Process.Kill(); err != nil {
+ t.Fatalf("SIGKILL instance child: %v", err)
+ }
+ <-proc.done
+ abandonCtx, abandonCancel := context.WithTimeout(context.Background(), 30*time.Second)
+ rep.Abandon(abandonCtx)
+ abandonCancel()
+ owner.shutdownOnce.Do(func() {}) // consume: cleanup must not re-drain a corpse
+ t.Logf("owner %s hard-killed: heartbeat left in S3 (stale after %s), lease left to expire (TTL %s)",
+ owner.name, survivor.mesh.staleAfter, activator.LeaseTTL)
+
+ // --- Act 3: the survivor takes over once staleness + lease expiry pass. ---
+ start := time.Now()
+ body := pollUntilServed(t, survivor.front.URL, host, 120*time.Second)
+ if !strings.Contains(body, groupName) {
+ t.Fatalf("survivor lost the data: %s does not contain %q", body, groupName)
+ }
+ if !survivor.act.entryAwake(host) {
+ t.Fatal("instance is not awake on the survivor")
+ }
+ t.Logf("survivor %s took over %s after the owner's death with data intact: %s",
+ survivor.name, time.Since(start).Truncate(time.Millisecond), body)
+}
diff --git a/mesh_test.go b/mesh_test.go
new file mode 100644
index 0000000..2929265
--- /dev/null
+++ b/mesh_test.go
@@ -0,0 +1,633 @@
+package main
+
+// Unit + in-process integration tests for the activator mesh (mesh.go): HRW
+// determinism and minimal redistribution, membership staleness, the
+// owner-vs-proxy routing decision, proxy header/Host semantics, drain
+// ordering (heartbeat first), the due-sweep ownership gate, and the mesh-off
+// tripwire. Two "boxes" here are two activator+router pairs sharing an
+// in-memory heartbeat store — the store IS the bucket, which is the whole
+// membership design. The real-S3, real-process version (proxy-to-owner and
+// the kill-a-box failover, drain and hard-kill variants) is in
+// mesh_live_test.go.
+
+import (
+ "context"
+ "fmt"
+ "io"
+ "net"
+ "net/http"
+ "net/http/httptest"
+ "net/http/httputil"
+ "net/url"
+ "path/filepath"
+ "sync"
+ "sync/atomic"
+ "testing"
+ "time"
+
+ "lchat/internal/activator"
+)
+
+// --- fakes ---------------------------------------------------------------------
+
+// fakeMeshStore is an in-memory stand-in for the S3 heartbeat prefix. Shared
+// between boxes it plays the bucket; rec (optional) feeds a fakeBackend's
+// order log so drain tests can pin event ordering.
+type fakeMeshStore struct {
+ mu sync.Mutex
+ beats map[string]time.Time
+ now func() time.Time
+ rec func(ev string)
+}
+
+func newFakeMeshStore(now func() time.Time) *fakeMeshStore {
+ return &fakeMeshStore{beats: make(map[string]time.Time), now: now}
+}
+
+func (s *fakeMeshStore) Put(_ context.Context, peer string) error {
+ s.mu.Lock()
+ s.beats[peer] = s.now()
+ rec := s.rec
+ s.mu.Unlock()
+ if rec != nil {
+ rec("hb-put")
+ }
+ return nil
+}
+
+func (s *fakeMeshStore) List(context.Context) ([]activator.Heartbeat, error) {
+ s.mu.Lock()
+ defer s.mu.Unlock()
+ out := make([]activator.Heartbeat, 0, len(s.beats))
+ for peer, seen := range s.beats {
+ out = append(out, activator.Heartbeat{Peer: peer, LastSeen: seen})
+ }
+ return out, nil
+}
+
+func (s *fakeMeshStore) Delete(_ context.Context, peer string) error {
+ s.mu.Lock()
+ delete(s.beats, peer)
+ rec := s.rec
+ s.mu.Unlock()
+ if rec != nil {
+ rec("hb-delete")
+ }
+ return nil
+}
+
+func (s *fakeMeshStore) setBeat(peer string, at time.Time) {
+ s.mu.Lock()
+ defer s.mu.Unlock()
+ s.beats[peer] = at
+}
+
+func (s *fakeMeshStore) hasBeat(peer string) bool {
+ s.mu.Lock()
+ defer s.mu.Unlock()
+ _, ok := s.beats[peer]
+ return ok
+}
+
+// hostOwnedBy generates deterministic host names until one's HRW owner is the
+// wanted box — tests can't hardcode hosts because box identities (httptest
+// ports) vary per run.
+func hostOwnedBy(boxes []string, owner string) string {
+ for i := 0; ; i++ {
+ h := fmt.Sprintf("space-%d.test", i)
+ if hrwOwner(h, boxes) == owner {
+ return h
+ }
+ }
+}
+
+// meshTestBox is one in-process "box": activator (fake backend) + router +
+// front door + a live unix-socket listener standing in for any woken instance
+// (it echoes the box's name and records what the backend saw).
+type meshTestBox struct {
+ name string
+ act *activatorSvc
+ b *fakeBackend
+ reg *Registry
+ rt *router
+ front *httptest.Server
+ mesh *meshSvc
+ sock string
+
+ seenHost atomic.Value // last Host the backend instance saw
+ fwdLeaked atomic.Bool // the forwarded marker reached the backend
+}
+
+func (box *meshTestBox) self() string { return box.front.URL }
+
+func newMeshTestBox(t *testing.T, name string, store meshStore, clk *fakeClock) *meshTestBox {
+ t.Helper()
+ box := &meshTestBox{name: name, b: &fakeBackend{}}
+
+ reg, err := openRegistry(filepath.Join(t.TempDir(), "registry.db"))
+ if err != nil {
+ t.Fatalf("open registry: %v", err)
+ }
+ box.reg = reg
+ box.act = newActivator(hibernateConfig{Bucket: "test-bucket", Idle: time.Minute}, box.b, reg)
+ box.act.now = clk.now
+
+ box.sock = tmpSock(t)
+ ln, err := net.Listen("unix", box.sock)
+ if err != nil {
+ t.Fatalf("listen unix: %v", err)
+ }
+ srv := &http.Server{Handler: http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+ box.seenHost.Store(r.Host)
+ if r.Header.Get(meshForwardedHeader) != "" {
+ box.fwdLeaked.Store(true)
+ }
+ io.WriteString(w, name)
+ })}
+ go srv.Serve(ln)
+ t.Cleanup(func() { srv.Close() })
+
+ box.rt, err = newRouter(reg, box.act)
+ if err != nil {
+ t.Fatalf("new router: %v", err)
+ }
+ box.front = httptest.NewServer(box.rt)
+ t.Cleanup(box.front.Close)
+
+ box.mesh = newMesh(box.front.URL, store)
+ box.mesh.now = clk.now
+ box.act.mesh = box.mesh
+ return box
+}
+
+// addHost registers a hibernating route for host on this box (pointing at its
+// stand-in instance socket) and reloads the router.
+func (box *meshTestBox) addHost(t *testing.T, host string) {
+ t.Helper()
+ if err := box.reg.upsert(Route{Host: host, Kind: "instance", Socket: box.sock, Hibernate: true}); err != nil {
+ t.Fatal(err)
+ }
+ if err := box.rt.reload(); err != nil {
+ t.Fatalf("reload: %v", err)
+ }
+}
+
+// getVia is proxyGet plus arbitrary headers.
+func getVia(t *testing.T, base, host string, hdr map[string]string) (string, int) {
+ t.Helper()
+ req, err := http.NewRequest(http.MethodGet, base+"/", nil)
+ if err != nil {
+ t.Fatalf("new request: %v", err)
+ }
+ req.Host = host
+ for k, v := range hdr {
+ req.Header.Set(k, v)
+ }
+ resp, err := http.DefaultClient.Do(req)
+ if err != nil {
+ t.Fatalf("request: %v", err)
+ }
+ defer resp.Body.Close()
+ b, _ := io.ReadAll(resp.Body)
+ return string(b), resp.StatusCode
+}
+
+// --- rendezvous hashing ----------------------------------------------------------
+
+// hrwOwner is a pure function: same answer for any ordering of the member
+// set, on every box, forever — that's what lets placement need no directory.
+func TestHRWOwnerDeterministic(t *testing.T) {
+ boxes := []string{"http://a", "http://b", "http://c"}
+ perms := [][]string{
+ {"http://a", "http://b", "http://c"},
+ {"http://c", "http://a", "http://b"},
+ {"http://b", "http://c", "http://a"},
+ }
+ owned := map[string]int{}
+ for i := 0; i < 300; i++ {
+ h := fmt.Sprintf("host-%d.test", i)
+ want := hrwOwner(h, boxes)
+ for _, p := range perms {
+ if got := hrwOwner(h, p); got != want {
+ t.Fatalf("owner of %s depends on member order: %q vs %q", h, got, want)
+ }
+ }
+ owned[want]++
+ }
+ // The hash must actually spread load — every box owns a healthy share.
+ for _, b := range boxes {
+ if owned[b] < 50 {
+ t.Fatalf("box %s owns only %d/300 hosts — distribution is broken: %v", b, owned[b], owned)
+ }
+ }
+ if hrwOwner("x.test", []string{"http://solo"}) != "http://solo" {
+ t.Fatal("single box does not own everything")
+ }
+ if hrwOwner("x.test", nil) != "" {
+ t.Fatal("empty member set should own nothing")
+ }
+}
+
+// The property that makes membership churn cheap: adding a box moves only the
+// ~1/N of hosts it steals (all TO the joiner), removing one moves only the
+// leaver's hosts (scattered over survivors) — everyone else stays put, so a
+// scale event never stampedes the fleet.
+func TestHRWRedistributionMinimal(t *testing.T) {
+ four := []string{"http://a", "http://b", "http://c", "http://d"}
+ five := append(append([]string{}, four...), "http://e")
+ const n = 1000
+
+ before := make(map[string]string, n)
+ for i := 0; i < n; i++ {
+ h := fmt.Sprintf("host-%d.test", i)
+ before[h] = hrwOwner(h, four)
+ }
+
+ // Join: every moved host must have moved onto the new box.
+ moved := 0
+ for h, was := range before {
+ now := hrwOwner(h, five)
+ if now != was {
+ moved++
+ if now != "http://e" {
+ t.Fatalf("join moved %s from %s to %s — only the joiner may steal", h, was, now)
+ }
+ }
+ }
+ if moved < n/10 || moved > n/2 {
+ t.Fatalf("join moved %d/%d hosts, want ~1/5 (a fifth box should steal about its share)", moved, n)
+ }
+
+ // Leave: only the leaver's hosts move.
+ three := []string{"http://a", "http://c", "http://d"}
+ for h, was := range before {
+ now := hrwOwner(h, three)
+ if was == "http://b" {
+ if now == "http://b" {
+ t.Fatalf("%s still owned by the removed box", h)
+ }
+ } else if now != was {
+ t.Fatalf("leave moved %s from %s to %s — hosts not owned by the leaver must not move", h, was, now)
+ }
+ }
+}
+
+// --- membership --------------------------------------------------------------
+
+// Staleness: a heartbeat older than staleAfter means a dead box — judged at
+// read time, so a box that dies between polls drops out on the clock, not on
+// the next LIST. Self is always a member (a box counts itself even before its
+// first PUT lands), and garbage records never enter the hash.
+func TestMeshMembershipStaleness(t *testing.T) {
+ clk := &fakeClock{t: time.Unix(1_700_000_000, 0)}
+ store := newFakeMeshStore(clk.now)
+ m := newMesh("http://a", store)
+ m.now = clk.now
+
+ // Before any beat: membership of one — the degenerate single box.
+ if got := m.liveBoxes(); len(got) != 1 || got[0] != "http://a" {
+ t.Fatalf("liveBoxes before any beat = %v, want just self", got)
+ }
+
+ store.setBeat("http://b", clk.now())
+ store.setBeat("this is not a url", clk.now()) // a foreign/garbage record
+ m.beat()
+ if !store.hasBeat("http://a") {
+ t.Fatal("beat did not PUT our own heartbeat")
+ }
+ live := m.liveBoxes()
+ if len(live) != 2 {
+ t.Fatalf("liveBoxes = %v, want self+b (garbage excluded)", live)
+ }
+
+ // b goes quiet: past staleAfter it is dead — with NO new list needed.
+ clk.advance(m.staleAfter + time.Second)
+ if got := m.liveBoxes(); len(got) != 1 || got[0] != "http://a" {
+ t.Fatalf("liveBoxes with a stale peer = %v, want just self", got)
+ }
+}
+
+// The routing decision: proxy only when the HRW owner is another LIVE box;
+// own it (wake locally) when the owner is us — or dead, because a stale owner
+// is out of the member set and the hash has already reassigned. The lease
+// arbitrates if this view is ever wrong; the decision only places.
+func TestMeshRoutingDecision(t *testing.T) {
+ clk := &fakeClock{t: time.Unix(1_700_000_000, 0)}
+ store := newFakeMeshStore(clk.now)
+ m := newMesh("http://a", store)
+ m.now = clk.now
+ store.setBeat("http://b", clk.now())
+ m.beat()
+
+ boxes := []string{"http://a", "http://b"}
+ theirs := hostOwnedBy(boxes, "http://b")
+ ours := hostOwnedBy(boxes, "http://a")
+
+ if peer, remote := m.remoteOwner(theirs); !remote || peer != "http://b" {
+ t.Fatalf("remoteOwner(%s) = %q, %v; want b, true", theirs, peer, remote)
+ }
+ if _, remote := m.remoteOwner(ours); remote {
+ t.Fatalf("remoteOwner(%s) says proxy a host we own", ours)
+ }
+ if m.owns(theirs) || !m.owns(ours) {
+ t.Fatal("owns() disagrees with remoteOwner()")
+ }
+
+ // The owner dies: its hosts fall through to a local wake.
+ clk.advance(m.staleAfter + time.Second)
+ if _, remote := m.remoteOwner(theirs); remote {
+ t.Fatal("remoteOwner still proxies to a stale (dead) box — its hosts must wake locally")
+ }
+ if !m.owns(theirs) {
+ t.Fatal("survivor does not own the dead box's host")
+ }
+}
+
+// --- the proxy hop -------------------------------------------------------------
+
+// Scheme/Host/SNI semantics of the peer hop: the instance hostname survives
+// (the peer routes by Host); for https peers the outgoing URL keeps the
+// instance host so SNI and cert verification run against a name the peer can
+// actually prove (its wildcard/per-host cert), while the dialer pins TCP to
+// the peer's address; for http peers the URL targets the peer directly. The
+// forwarded marker rides every hop. InsecureSkipVerify appears nowhere.
+func TestPeerProxyRewrite(t *testing.T) {
+ rewrite := func(p *httputil.ReverseProxy, host string) *http.Request {
+ in := httptest.NewRequest("GET", "http://placeholder/api/x", nil)
+ in.Host = host
+ in.RemoteAddr = "192.0.2.7:4711"
+ out := in.Clone(context.Background())
+ p.Rewrite(&httputil.ProxyRequest{In: in, Out: out})
+ return out
+ }
+
+ httpsPeer, _ := url.Parse("https://box-b.internal")
+ out := rewrite(newPeerProxy("http://self", httpsPeer), "space.11ch.at")
+ if out.URL.Scheme != "https" || out.URL.Host != "space.11ch.at" || out.Host != "space.11ch.at" {
+ t.Fatalf("https hop: scheme=%s urlHost=%s host=%s — SNI/verification must target the instance host", out.URL.Scheme, out.URL.Host, out.Host)
+ }
+ if got := peerDialAddr(httpsPeer); got != "box-b.internal:443" {
+ t.Fatalf("https dial addr = %q, want the peer's address", got)
+ }
+ if out.Header.Get(meshForwardedHeader) != "http://self" {
+ t.Fatal("forwarded marker missing — the one-hop loop guard would not hold")
+ }
+
+ httpPeer, _ := url.Parse("http://10.0.0.5:9000")
+ out = rewrite(newPeerProxy("http://self", httpPeer), "space.11ch.at")
+ if out.URL.Scheme != "http" || out.URL.Host != "10.0.0.5:9000" || out.Host != "space.11ch.at" {
+ t.Fatalf("http hop: scheme=%s urlHost=%s host=%s — dial the peer, Host stays the instance", out.URL.Scheme, out.URL.Host, out.Host)
+ }
+ if got := peerDialAddr(httpPeer); got != "10.0.0.5:9000" {
+ t.Fatalf("http dial addr = %q", got)
+ }
+ bare, _ := url.Parse("http://box-c.internal")
+ if got := peerDialAddr(bare); got != "box-c.internal:80" {
+ t.Fatalf("default http port: %q", got)
+ }
+}
+
+// Two boxes, one shared "bucket": a request landing on the non-owner is
+// reverse-proxied to the owner (which wakes the instance), the instance
+// hostname survives the hop, the forwarded marker never reaches a backend,
+// a request already carrying the marker is served where it stands (one hop
+// max), and a host with a live local entry stays local even when the hash
+// points elsewhere.
+func TestMeshProxyToOwner(t *testing.T) {
+ clk := &fakeClock{t: time.Unix(1_700_000_000, 0)}
+ store := newFakeMeshStore(clk.now)
+ box1 := newMeshTestBox(t, "box1", store, clk)
+ box2 := newMeshTestBox(t, "box2", store, clk)
+ // Beat both (twice for box1, whose first beat predates box2's PUT) so
+ // each box sees the full membership.
+ box1.mesh.beat()
+ box2.mesh.beat()
+ box1.mesh.beat()
+
+ boxes := []string{box1.self(), box2.self()}
+ host := hostOwnedBy(boxes, box2.self())
+ box1.addHost(t, host)
+ box2.addHost(t, host)
+
+ // (a) Request to the NON-owner: proxied to the owner and served there.
+ if body, code := proxyGet(t, box1.front.URL, host); code != 200 || body != "box2" {
+ t.Fatalf("non-owner request: got %d %q, want 200 box2 (proxied to owner)", code, body)
+ }
+ if acq, _, _ := box2.b.counts(); acq != 1 {
+ t.Fatalf("owner acquires = %d, want 1 (the wake happened on the owner)", acq)
+ }
+ if acq, _, starts := box1.b.counts(); acq != 0 || starts != 0 {
+ t.Fatalf("non-owner touched its own activator (acquires=%d starts=%d)", acq, starts)
+ }
+ if got, _ := box2.seenHost.Load().(string); got != host {
+ t.Fatalf("owner backend saw Host %q, want %q — the hostname must survive the hop", got, host)
+ }
+ if box2.fwdLeaked.Load() {
+ t.Fatal("the forwarded marker leaked through to the backend")
+ }
+
+ // (b) Request straight to the owner: local, no extra wake.
+ if body, code := proxyGet(t, box2.front.URL, host); code != 200 || body != "box2" {
+ t.Fatalf("owner request: got %d %q", code, body)
+ }
+ if acq, _, _ := box2.b.counts(); acq != 1 {
+ t.Fatalf("owner re-acquired for a host it already has awake (acquires=%d)", acq)
+ }
+
+ // (c) One hop max: a request already carrying the forwarded marker is
+ // served where it stands — here the non-owner, which wakes it locally
+ // (the lease would arbitrate a real conflict; the fake just observes).
+ var host2 string // a second box2-owned host, distinct from the first
+ for i := 0; ; i++ {
+ if h := fmt.Sprintf("space-b-%d.test", i); hrwOwner(h, boxes) == box2.self() {
+ host2 = h
+ break
+ }
+ }
+ box1.addHost(t, host2)
+ box2.addHost(t, host2)
+ if body, code := getVia(t, box1.front.URL, host2, map[string]string{meshForwardedHeader: "http://elsewhere"}); code != 200 || body != "box1" {
+ t.Fatalf("forwarded request: got %d %q, want 200 box1 (served in place, never re-proxied)", code, body)
+ }
+ if acq, _, _ := box1.b.counts(); acq != 1 {
+ t.Fatalf("non-owner acquires = %d after forwarded request, want 1", acq)
+ }
+
+ // (d) Live-local-entry rule: host2 is now awake on box1, so an ordinary
+ // request to box1 stays local despite the hash pointing at box2 — an
+ // awake instance finishes its life where it is.
+ if body, code := proxyGet(t, box1.front.URL, host2); code != 200 || body != "box1" {
+ t.Fatalf("live-entry request: got %d %q, want 200 box1 (local entry wins over the hint)", code, body)
+ }
+ if acq, _, _ := box2.b.counts(); acq != 1 {
+ t.Fatalf("owner was woken for a host living on its peer (acquires=%d)", acq)
+ }
+}
+
+// --- mesh off: the tripwire ------------------------------------------------------
+
+// With the mesh off (LCHAT_MESH_SELF unset → act.mesh nil) the activator is
+// byte-identical to the single-box shape: a host whose HRW owner WOULD be a
+// peer is woken locally, nothing is proxied, and no heartbeat is ever
+// written. The same box with only the mesh flipped on proxies the same-shaped
+// request — proving the nil check is the single divergence point.
+func TestMeshOffTripwire(t *testing.T) {
+ clk := &fakeClock{t: time.Unix(1_700_000_000, 0)}
+ store := newFakeMeshStore(clk.now)
+ box1 := newMeshTestBox(t, "box1", store, clk)
+ box2 := newMeshTestBox(t, "box2", store, clk)
+ box1.act.mesh = nil // the tripwire: box1 is mesh-off
+ box2.mesh.beat() // box2 participates; its heartbeat is in the store
+
+ boxes := []string{box1.self(), box2.self()}
+ host := hostOwnedBy(boxes, box2.self())
+ box1.addHost(t, host)
+ box2.addHost(t, host)
+
+ if body, code := proxyGet(t, box1.front.URL, host); code != 200 || body != "box1" {
+ t.Fatalf("mesh-off request: got %d %q, want 200 box1 (woken locally, never proxied)", code, body)
+ }
+ if acq, _, _ := box1.b.counts(); acq != 1 {
+ t.Fatalf("mesh-off box acquires = %d, want 1", acq)
+ }
+ if acq, _, _ := box2.b.counts(); acq != 0 {
+ t.Fatalf("peer was touched by a mesh-off box (acquires=%d)", acq)
+ }
+ if store.hasBeat(box1.self()) {
+ t.Fatal("mesh-off box wrote a heartbeat")
+ }
+
+ // Contrast: flip the mesh on — the ONLY change — and the same-shaped
+ // request (a fresh host, since the first now has a live local entry)
+ // proxies to the owner.
+ box1.act.mesh = box1.mesh
+ box1.mesh.beat()
+ var host2 string
+ for i := 0; ; i++ {
+ h := fmt.Sprintf("contrast-%d.test", i)
+ if hrwOwner(h, boxes) == box2.self() {
+ host2 = h
+ break
+ }
+ }
+ box1.addHost(t, host2)
+ box2.addHost(t, host2)
+ if body, code := proxyGet(t, box1.front.URL, host2); code != 200 || body != "box2" {
+ t.Fatalf("mesh-on contrast: got %d %q, want 200 box2 (proxied)", code, body)
+ }
+}
+
+// meshFromEnv: unset = off (nil, nil — the zero-change default); set without
+// the hibernation bucket = a loud config error; malformed URLs rejected.
+func TestMeshFromEnv(t *testing.T) {
+ t.Setenv("LCHAT_MESH_SELF", "")
+ if m, err := meshFromEnv(&hibernateConfig{Bucket: "b"}); err != nil || m != nil {
+ t.Fatalf("unset: got %v, %v; want nil, nil (mesh off)", m, err)
+ }
+
+ t.Setenv("LCHAT_MESH_SELF", "https://box-a.internal")
+ if _, err := meshFromEnv(nil); err == nil {
+ t.Fatal("mesh without a hibernation bucket should be a config error")
+ }
+
+ for _, bad := range []string{"box-a.internal", "ftp://box-a", "https://box-a/path"} {
+ t.Setenv("LCHAT_MESH_SELF", bad)
+ if _, err := meshFromEnv(&hibernateConfig{Bucket: "b", Region: "eu-west-1"}); err == nil {
+ t.Fatalf("LCHAT_MESH_SELF %q should be rejected", bad)
+ }
+ }
+
+ t.Setenv("LCHAT_MESH_SELF", "https://box-a.internal/")
+ m, err := meshFromEnv(&hibernateConfig{Bucket: "b", Region: "eu-west-1"})
+ if err != nil || m == nil {
+ t.Fatalf("valid self: %v, %v", m, err)
+ }
+ if m.self != "https://box-a.internal" {
+ t.Fatalf("self = %q, want the trailing slash trimmed", m.self)
+ }
+}
+
+// --- drain ordering ---------------------------------------------------------------
+
+// The drain contract, order-pinned: the heartbeat is deleted FIRST — before
+// the next-due probe, before the child stops, before the final checkpoint and
+// lease release — so peers take over the moment the leases free up. Same code
+// path for deploys, box failure drills, and scale-down.
+func TestMeshDrainDeletesHeartbeatFirst(t *testing.T) {
+ a, b, clk, _ := testActivator(t, time.Hour)
+ store := newFakeMeshStore(clk.now)
+ store.rec = b.record // interleave heartbeat ops with the backend's order log
+ m := newMesh("http://box-a", store)
+ m.now = clk.now
+ m.interval = time.Hour // the loop must never beat on its own mid-test
+ a.mesh = m
+ m.start()
+
+ release, err := a.beginRequest(context.Background(), testRoute("drain.example.com"))
+ if err != nil {
+ t.Fatalf("beginRequest: %v", err)
+ }
+ release()
+
+ ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
+ defer cancel()
+ a.shutdown(ctx)
+
+ if store.hasBeat("http://box-a") {
+ t.Fatal("drain left our heartbeat behind")
+ }
+ ev := b.events()
+ del := indexOf(ev, "hb-delete")
+ if del == -1 {
+ t.Fatalf("no heartbeat delete in %v", ev)
+ }
+ if put := indexOf(ev, "hb-put"); put == -1 || put > del {
+ t.Fatalf("heartbeat lifecycle out of order: %v", ev)
+ }
+ for _, later := range []string{"nextDue", "stop:instance"} {
+ if i := indexOf(ev, later); i == -1 || i < del {
+ t.Fatalf("drain order wrong: %q at %d, hb-delete at %d — the heartbeat must go first (%v)", later, i, del, ev)
+ }
+ }
+}
+
+// --- due-time sweeps under the mesh -----------------------------------------------
+
+// Only the HRW owner due-wakes a host; a non-owner skips it every tick. When
+// the owner dies, staleness reassigns ownership and the next sweep fires the
+// wake here — failover for schedules is the same eventual path as requests.
+func TestMeshDueSweepOnlyOwner(t *testing.T) {
+ a, b, clk, reg := testActivator(t, 10*time.Second)
+ store := newFakeMeshStore(clk.now)
+ m := newMesh("http://box-a", store)
+ m.now = clk.now
+ a.mesh = m
+ store.setBeat("http://box-b", clk.now())
+ m.beat()
+
+ host := hostOwnedBy([]string{"http://box-a", "http://box-b"}, "http://box-b")
+ route := testRoute(host)
+ if err := reg.upsert(route); err != nil {
+ t.Fatal(err)
+ }
+ if err := reg.setNextDue(host, clk.now().Add(-time.Second).Unix()); err != nil {
+ t.Fatal(err)
+ }
+
+ // The peer owns it: our sweeps must not wake it, however overdue.
+ for i := 0; i < 3; i++ {
+ a.sweepOnce()
+ }
+ if _, _, starts := b.counts(); starts != 0 {
+ t.Fatalf("starts=%d — a non-owner due-woke a host", starts)
+ }
+
+ // The owner dies (heartbeat goes stale): ownership reassigns to us on the
+ // next read, and the sweep fires the wake.
+ clk.advance(m.staleAfter + time.Second)
+ a.sweepOnce()
+ waitFor(t, "post-failover due wake", func() bool { return a.entryAwake(host) })
+ if _, _, starts := b.counts(); starts != 1 {
+ t.Fatalf("starts=%d after owner death, want 1", starts)
+ }
+}
diff --git a/router.go b/router.go
index 56503ee..d8ba225 100644
--- a/router.go
+++ b/router.go
@@ -162,8 +162,11 @@ func unixProxy(socketPath string) *httputil.ReverseProxy {
pr.Out.Host = pr.In.Host // backend sees the real hostname
// The sidecar marker is a router-edge signal only (activity
// exclusion, activator.go); strip it so nothing downstream can
- // ever grow a meaning an outsider could spoof.
+ // ever grow a meaning an outsider could spoof. The mesh's
+ // forwarded marker (one-hop loop guard, mesh.go) gets the same
+ // treatment for the same reason.
pr.Out.Header.Del(bot.SidecarHeader)
+ pr.Out.Header.Del(meshForwardedHeader)
},
Transport: &http.Transport{
DialContext: func(ctx context.Context, _, _ string) (net.Conn, error) {
@@ -231,13 +234,31 @@ func runRouter(args []string) {
// Unset keeps the classic shape — instances under systemd, router as a
// pure proxy — so self-hosters and the current fleet are untouched.
var act *activatorSvc
- if hc, err := hibernateConfigFromEnv(); err != nil {
+ hc, err := hibernateConfigFromEnv()
+ if err != nil {
+ log.Fatalf("router: %v", err)
+ }
+ // The activator mesh (mesh.go): opt-in via LCHAT_MESH_SELF, rides on the
+ // hibernation bucket (meshFromEnv errors if it's set without one). Unset =
+ // single box, no mesh codepaths.
+ mesh, err := meshFromEnv(hc)
+ if err != nil {
log.Fatalf("router: %v", err)
- } else if hc != nil {
+ }
+ if hc != nil {
// The registry doubles as the due-time index's store (routes.next_due),
// so scheduled deliveries survive router restarts and fire on time even
// for sleeping instances.
act = newActivator(*hc, &s3Backend{cfg: *hc}, reg)
+ if mesh != nil {
+ // Wired and beating before act.start(): the sweeper consults the
+ // mesh from its first tick, and the first request deserves a
+ // membership view.
+ act.mesh = mesh
+ mesh.start()
+ log.Printf("router: activator mesh on — this box is %s (heartbeats s3://%s/%s every %s, peers stale after %s)",
+ mesh.self, hc.Bucket, hc.meshPrefix(), mesh.interval, mesh.staleAfter)
+ }
act.start()
log.Printf("router: hibernation active (bucket %s, prefix %q, idle %s) — flagged instances wake on demand",
hc.Bucket, hc.Prefix, hc.Idle)