eleven / messenger Public

server: per-plan replication classes on the activator's S3 replica (stream/batched/daily) (#866)

* server: per-plan replication classes on the activator's S3 replica (stream/batched/daily)

The durability ladder's mechanism half: each hibernating host carries a
replication class in the registry (routes.replication) that sets litestream's
replica sync interval at wake — stream (change-driven ~1s, the status quo and
the unset default), batched (at most every 5m), daily (wake + every 24h).
Every graceful sleep/deploy/shutdown still flushes in full via SyncAndWait, so
the class only bounds what a HARD box crash can lose while awake — and its S3
Tier-1 request bill, which is the point: streaming's change-driven PUTs are
the one per-instance cost that scales with activity, and it lands on spaces
whose plans can't carry it.

- registry: routes.replication column (additive migration), narrow validated
  setReplication, class constants + replicationSyncInterval mapping
- activator: doWake threads the route's class into replicate(); the actLease
  seam and internal/activator.Replicate take the interval (0 = today's default)
- CLI: lchat replication <host> [class] to read/set (+ router reload; applies
  at next wake), REPLICATION column in lchat instances, add-instance
  --replication for birth-time classing (what the provisioner will use)

Billing→class mapping (cloud/controld push) and docs land separately.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>

* server: replication class rides the mesh route payload

A space that fails over to a peer box must keep its durability rung — without
this the class silently reverts to stream on the peer (safe for the data,
wrong for the plan's economics). upsert deliberately skips class validation so
an older box can store a newer peer's unknown class verbatim and fail-safe to
stream until it updates.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>

---------

Co-authored-by: Claude Fable 5 <noreply@anthropic.com>

Paul Campbell 3 months ago 194dae5b67b4209748f6779fc4dbbc871135a5e4 parent cbfa92e

9 files changed, +377 −30

activator.go +8 −6
activator_test.go +49 −4
control.go +101 −6
control_test.go +115 −0
internal/activator/replica.go +18 −5
main.go +3 −0
meshreplicate.go +3 −1
meshreplicate_test.go +7 −0
registry.go +73 −8

diff --git a/activator.go b/activator.go

index 53e4bc5..ced1a4b 100644

--- a/activator.go

+++ b/activator.go

@@ -162,8 +162,9 @@ type actDueStore interface {

// actLease is a held-but-not-yet-replicating lease (wake in progress).

type actLease interface {

// replicate trades the lease for running replication of the (now present)

- // local DB; the lease handle is spent on success.

- replicate(ctx context.Context, dbPath string) (actRep, error)

+ // local DB; the lease handle is spent on success. syncInterval is the

+ // route's replication-class tick (replicationSyncInterval; 0 = stream).

+ replicate(ctx context.Context, dbPath string, syncInterval time.Duration) (actRep, error)

// release gives the lease up — the bail-out when a wake fails partway.

release(ctx context.Context) error

}

@@ -622,8 +623,9 @@ func (a *activatorSvc) doWake(ctx context.Context, rt Route) (*wokenInstance, er

return nil, fmt.Errorf("start %s: %w", host, err)

}

- // 4. Trade the held lease for running replication.

- rep, err := lease.replicate(ctx, dbPath)

+ // 4. Trade the held lease for running replication, at the route's

+ // replication-class sync interval (the per-plan durability ladder).

+ rep, err := lease.replicate(ctx, dbPath, replicationSyncInterval(rt.Replication))

if err != nil {

stopCtx, cancel := context.WithTimeout(context.Background(), actStopTimeout)

defer cancel()

@@ -1075,8 +1077,8 @@ func (b *s3Backend) acquire(ctx context.Context, host string) (actLease, error)

type s3Lease struct{ held *activator.HeldLease }

-func (l *s3Lease) replicate(ctx context.Context, dbPath string) (actRep, error) {

- return l.held.Replicate(ctx, dbPath)

+func (l *s3Lease) replicate(ctx context.Context, dbPath string, syncInterval time.Duration) (actRep, error) {

+ return l.held.Replicate(ctx, dbPath, syncInterval)

}

func (l *s3Lease) release(ctx context.Context) error { return l.held.Release(ctx) }

diff --git a/activator_test.go b/activator_test.go

index 9f4ec0d..869c0ed 100644

--- a/activator_test.go

+++ b/activator_test.go

@@ -93,10 +93,20 @@ func (p *fakeProc) isStopped() bool {

return p.stopped

}

-type fakeLease struct{ rep *fakeRep }

+type fakeLease struct {

+ rep *fakeRep

+ b *fakeBackend // records the sync interval the wake threads through

-func (l *fakeLease) replicate(context.Context, string) (actRep, error) { return l.rep, nil }

-func (l *fakeLease) release(context.Context) error { return nil }

+func (l *fakeLease) replicate(_ context.Context, _ string, syncInterval time.Duration) (actRep, error) {

+ if l.b != nil {

+ l.b.mu.Lock()

+ l.b.syncIntervals = append(l.b.syncIntervals, syncInterval)

+ l.b.mu.Unlock()

+ }

+ return l.rep, nil

+func (l *fakeLease) release(context.Context) error { return nil }

type fakeBackend struct {

mu sync.Mutex

@@ -107,6 +117,9 @@ type fakeBackend struct {

startGate chan struct{} // when non-nil, start() blocks until closed

reps []*fakeRep

procs []*fakeProc

+ // syncIntervals records, per wake, the replication-class tick doWake

+ // threads into replicate() — the ladder's seam assertion.

+ syncIntervals []time.Duration

dueVal int64 // what the instance reports at the sleep-time next-due probe

dueErr error // when non-nil, the probe fails (dead socket, missing token, …)

@@ -152,7 +165,7 @@ func (b *fakeBackend) acquire(context.Context, string) (actLease, error) {

if err != nil {

return nil, err

}

- return &fakeLease{rep: rep}, nil

+ return &fakeLease{rep: rep, b: b}, nil

}

func (b *fakeBackend) restore(context.Context, string, string) error {

@@ -341,6 +354,38 @@ func TestActivatorWakeCoalesces(t *testing.T) {

}

+// The wake threads the route's replication class into replicate() as the

+// litestream sync interval — the per-plan durability ladder's seam. Unset

+// (every pre-ladder route) must thread 0, i.e. litestream's default: a deploy

+// of the ladder changes no host's durability by itself.

+func TestActivatorWakeThreadsReplicationClass(t *testing.T) {

+ a, b, _, _ := testActivator(t, time.Minute)

+ for i, tc := range []struct {

+ class string

+ want time.Duration

+ }{

+ {"", 0},

+ {replicationStream, 0},

+ {replicationBatched, 5 * time.Minute},

+ {replicationDaily, 24 * time.Hour},

+ } {

+ rt := testRoute(fmt.Sprintf("ladder%d.example.com", i))

+ rt.Replication = tc.class

+ release, err := a.beginRequest(context.Background(), rt)

+ if err != nil {

+ t.Fatalf("beginRequest(%q): %v", tc.class, err)

+ }

+ release()

+ b.mu.Lock()

+ got := b.syncIntervals[len(b.syncIntervals)-1]

+ b.mu.Unlock()

+ if got != tc.want {

+ t.Errorf("class %q threaded sync interval %v, want %v", tc.class, got, tc.want)

+ }

// After the idle window with nothing in flight, the sweeper hibernates the

// instance: process stopped, replication closed (final checkpoint + lease

// release), entry forgotten; the next request wakes it afresh.

diff --git a/control.go b/control.go

index 8894013..63f16e3 100644

--- a/control.go

+++ b/control.go

@@ -127,9 +127,14 @@ type addInstanceOpts struct {

// Hibernate registers the instance as activator-managed (#550 Part 4 v2):

// no systemd unit is created — the router wakes it on first request. Needs

// the router running with LCHAT_HIBERNATE_BUCKET, or the host never answers.

- Hibernate bool

- UnitTmpl string // default "lchat@"

- RouterUnit string // default "lchat-router"

+ Hibernate bool

+ // Replication is the durability class on the activator's S3 replica

+ // (registry.go: stream/batched/daily; "" = stream) — set at birth so a

+ // provisioned space starts on its plan's rung instead of the most

+ // expensive one. Inert unless Hibernate.

+ Replication string

+ UnitTmpl string // default "lchat@"

+ RouterUnit string // default "lchat-router"

}

// addInstanceResult is what a successful creation yields. SetupLink may be empty

@@ -191,10 +196,16 @@ func registerInstance(reg *Registry, opts addInstanceOpts) (Route, string, error

}

+ if !validReplication(opts.Replication) {

+ return Route{}, "", fmt.Errorf("unknown replication class %q (want %s, %s or %s)",

+ opts.Replication, replicationStream, replicationBatched, replicationDaily)

+ }

route := Route{Host: host, Kind: "instance", Socket: socket,

DBPath: filepath.Join(instanceDataDir, host+".db"), Status: "creating",

AppStoreURL: opts.AppStoreURL, HomeOrigin: opts.HomeOrigin, PushRelayURL: opts.PushRelayURL,

- HomeRegion: opts.HomeRegion, Provider: opts.Provider, Hibernate: opts.Hibernate}

+ HomeRegion: opts.HomeRegion, Provider: opts.Provider, Hibernate: opts.Hibernate,

+ Replication: opts.Replication}

if err := reg.upsert(route); err != nil {

return Route{}, "", fmt.Errorf("register: %v", err)

}

@@ -350,6 +361,7 @@ func runAddInstance(args []string) {

pushRelayURL := fs.String("push-relay-url", "", "per-instance LCHAT_PUSH_RELAY_URL")

pushRelayToken := fs.String("push-relay-token", "", "per-instance LCHAT_PUSH_RELAY_TOKEN")

hibernate := fs.Bool("hibernate", false, "register as activator-managed (no systemd unit; the router wakes it on demand — requires the router to run with LCHAT_HIBERNATE_BUCKET)")

+ replication := fs.String("replication", "", "replication class on the activator's S3 replica: stream (default), batched, or daily — the per-plan durability ladder; inert without --hibernate")

_ = fs.Parse(reorderFlagsFirst(fs, args))

reg, err := openRegistry(*registryPath)

@@ -364,6 +376,7 @@ func runAddInstance(args []string) {

PushRelayURL: *pushRelayURL,

PushRelayToken: *pushRelayToken,

Hibernate: *hibernate,

+ Replication: strings.ToLower(strings.TrimSpace(*replication)),

UnitTmpl: *unitTmpl,

RouterUnit: *routerUnit,

})

@@ -578,7 +591,7 @@ func runInstances(args []string) {

fmt.Println("(no routes)")

return

}

- fmt.Printf("%-34s %-9s %-9s %-12s %-9s %s\n", "HOST", "KIND", "STATUS", "VERSION", "HIBERNATE", "SOCKET")

+ fmt.Printf("%-34s %-9s %-9s %-12s %-9s %-11s %s\n", "HOST", "KIND", "STATUS", "VERSION", "HIBERNATE", "REPLICATION", "SOCKET")

for _, r := range routes {

ver := r.Version

if ver == "" {

@@ -590,13 +603,95 @@ func runInstances(args []string) {

if r.Hibernate {

hib = "yes"

}

- fmt.Printf("%-34s %-9s %-9s %-12s %-9s %s\n", r.Host, r.Kind, r.Status, ver, hib, r.Socket)

+ // Replication class: hibernating hosts show their effective class

+ // (unset = stream); a class staged on a systemd-managed host shows too

+ // (it's inert but deliberate); "-" = nothing set, nothing replicating.

+ repl := r.Replication

+ if repl == "" {

+ if r.Hibernate {

+ repl = replicationStream

+ } else {

+ repl = "-"

+ }

+ fmt.Printf("%-34s %-9s %-9s %-12s %-9s %-11s %s\n", r.Host, r.Kind, r.Status, ver, hib, repl, r.Socket)

if r.LastError != "" {

fmt.Printf(" last error: %s\n", r.LastError)

}

+// runReplication: `lchat replication <host> [class]` — read or set one

+// instance's replication class on the activator's S3 replica (the per-plan

+// durability ladder: stream / batched / daily; registry.go has the intervals).

+// The class is read at wake time, so a set takes effect on the host's next

+// wake — an awake instance keeps its current tick until it sleeps. Setting a

+// class on a systemd-managed (non-hibernating) host is allowed but inert: it

+// pre-stages the class for a later hibernate flip.

+func runReplication(args []string) {

+ fs := flag.NewFlagSet("replication", flag.ExitOnError)

+ registryPath := fs.String("registry", envOr("LCHAT_REGISTRY", "/data/registry.db"), "router registry SQLite path")

+ routerUnit := fs.String("router-unit", envOr("LCHAT_ROUTER_UNIT", "lchat-router"), "router systemd unit to reload")

+ _ = fs.Parse(reorderFlagsFirst(fs, args))

+ host := strings.ToLower(strings.TrimSpace(fs.Arg(0)))

+ if !validHost(host) {

+ die("replication: invalid host %q (usage: lchat replication <host> [stream|batched|daily])", host)

+ }

+ reg, err := openRegistry(*registryPath)

+ if err != nil {

+ die("replication: open registry: %v", err)

+ }

+ rt, err := reg.get(host)

+ if err != nil {

+ die("replication: %v", err)

+ }

+ if rt == nil {

+ die("replication: unknown host %q", host)

+ }

+ if rt.Kind != "instance" {

+ die("replication: %q is a %s route, not an instance", host, rt.Kind)

+ }

+ effective := func(class string) string {

+ if class == "" {

+ return replicationStream

+ }

+ return class

+ }

+ describe := map[string]string{

+ replicationStream: "change-driven S3 sync (~1s) — a hard crash loses seconds",

+ replicationBatched: "S3 sync at most every 5m — a hard crash loses minutes",

+ replicationDaily: "S3 sync at wake, then at most every 24h — a hard crash loses up to a day",

+ }

+ class := strings.ToLower(strings.TrimSpace(fs.Arg(1)))

+ if class == "" {

+ fmt.Printf("%s: %s (%s)\n", host, effective(rt.Replication), describe[effective(rt.Replication)])

+ if !rt.Hibernate {

+ fmt.Println(" note: this host is systemd-managed (not hibernating) — the class is inert; the box-wide litestream replicates it")

+ }

+ return

+ }

+ if err := reg.setReplication(host, class); err != nil {

+ die("replication: %v", err)

+ }

+ fmt.Printf("%s replication → %s: %s\n", host, class, describe[class])

+ fmt.Println(" every graceful sleep/deploy/shutdown still flushes in full — the class only bounds hard-crash loss while awake")

+ if !rt.Hibernate {

+ fmt.Println(" note: this host is systemd-managed (not hibernating) — the class is stored but inert until a hibernate flip")

+ return

+ }

+ // The router's route map only swaps on reload; without it the new class

+ // never reaches a wake. The registry write above is durable either way, so

+ // a failed reload is a warning (reload by hand), not a dead flip.

+ if err := systemctl("reload", *routerUnit); err != nil {

+ fmt.Printf(" WARNING: reload %s failed (%v) — the class is stored but inert until the router reloads\n", *routerUnit, err)

+ return

+ }

+ fmt.Println(" takes effect at the next wake (an awake instance keeps its current tick until it sleeps)")

// runHibernate: `lchat hibernate <host> --on|--off` — flip one instance between

// systemd-managed (always-on) and activator-managed (hibernating). Per-host by

// design: a production canary can hibernate while the rest of the fleet stays

diff --git a/control_test.go b/control_test.go

index a1b996e..d3f8276 100644

--- a/control_test.go

+++ b/control_test.go

@@ -431,6 +431,121 @@ INSERT INTO routes (host, kind, socket, created_at) VALUES ('old.example.com', '

}

+// A registry created before the replication column existed must gain it on

+// open, with existing rows reading as "" — the stream class, i.e. exactly the

+// pre-ladder behavior, so a deploy over the live fleet's registry changes no

+// host's durability until an operator (or billing) says so.

+func TestRegistryMigratesReplicationColumn(t *testing.T) {

+ path := filepath.Join(t.TempDir(), "reg.db")

+ // The schema as of the hibernate flag — everything current EXCEPT replication.

+ old, err := sql.Open("sqlite", path)

+ if err != nil {

+ t.Fatal(err)

+ }

+ if _, err := old.Exec(`

+CREATE TABLE routes (

+ host TEXT PRIMARY KEY,

+ kind TEXT NOT NULL,

+ socket TEXT NOT NULL,

+ status TEXT NOT NULL DEFAULT 'ready',

+ last_error TEXT NOT NULL DEFAULT '',

+ db_path TEXT NOT NULL DEFAULT '',

+ version TEXT NOT NULL DEFAULT '',

+ prev_version TEXT NOT NULL DEFAULT '',

+ app_store_url TEXT NOT NULL DEFAULT '',

+ home_origin TEXT NOT NULL DEFAULT '',

+ push_relay_url TEXT NOT NULL DEFAULT '',

+ push_relay_token TEXT NOT NULL DEFAULT '',

+ home_region TEXT NOT NULL DEFAULT '',

+ provider TEXT NOT NULL DEFAULT '',

+ hibernate INTEGER NOT NULL DEFAULT 0,

+ next_due INTEGER NOT NULL DEFAULT 0,

+ created_at INTEGER NOT NULL

+);

+INSERT INTO routes (host, kind, socket, hibernate, created_at) VALUES ('old.example.com', 'instance', '/run/lchat/old.sock', 1, 1700000000);

+`); err != nil {

+ t.Fatal(err)

+ }

+ if err := old.Close(); err != nil {

+ t.Fatal(err)

+ }

+ reg, err := openRegistry(path)

+ if err != nil {

+ t.Fatalf("open over pre-replication registry: %v", err)

+ }

+ rt, err := reg.get("old.example.com")

+ if err != nil || rt == nil {

+ t.Fatalf("get pre-existing row: %+v, %v", rt, err)

+ }

+ if rt.Replication != "" {

+ t.Errorf("pre-existing row: replication=%q, want \"\" (the stream default)", rt.Replication)

+ }

+ // The class round-trips through the narrow setter, an upsert carries it,

+ // and setReplication stays orthogonal to the rest of the route.

+ if err := reg.setReplication("old.example.com", replicationDaily); err != nil {

+ t.Fatalf("setReplication: %v", err)

+ }

+ if got, _ := reg.get("old.example.com"); got == nil || got.Replication != replicationDaily || !got.Hibernate {

+ t.Fatalf("after setReplication, route = %+v, want daily + hibernate intact", got)

+ }

+ rt, _ = reg.get("old.example.com")

+ rt.Version = "v2"

+ if err := reg.upsert(*rt); err != nil {

+ t.Fatalf("upsert rewrite: %v", err)

+ }

+ if got, _ := reg.get("old.example.com"); got == nil || got.Replication != replicationDaily {

+ t.Fatalf("route rewrite dropped the class: %+v", got)

+ }

+ // Garbage never lands: no write path may store a class the wake path would

+ // silently read as stream.

+ if err := reg.setReplication("old.example.com", "hourly"); err == nil {

+ t.Error("setReplication accepted an unknown class")

+ }

+ if got, _ := reg.get("old.example.com"); got == nil || got.Replication != replicationDaily {

+ t.Fatalf("rejected set still changed the row: %+v", got)

+ }

+ // An unknown host is a quiet no-op, matching setHibernate/setNextDue.

+ if err := reg.setReplication("ghost.example.com", replicationBatched); err != nil {

+ t.Fatalf("setReplication unknown host: %v", err)

+ }

+ // migrate is idempotent — reopening must not trip on the added column.

+ if _, err := openRegistry(path); err != nil {

+ t.Errorf("reopen after migration: %v", err)

+ }

+// The class→interval mapping is the ladder's contract: stream (and the unset

+// "" every pre-ladder registry row reads as) keeps litestream's default, and

+// the coarser classes stretch it — never the other way around.

+func TestReplicationSyncInterval(t *testing.T) {

+ for class, want := range map[string]time.Duration{

+ "": 0,

+ replicationStream: 0,

+ replicationBatched: 5 * time.Minute,

+ replicationDaily: 24 * time.Hour,

+ } {

+ if got := replicationSyncInterval(class); got != want {

+ t.Errorf("replicationSyncInterval(%q) = %v, want %v", class, got, want)

+ }

+ for _, ok := range []string{"", "stream", "batched", "daily"} {

+ if !validReplication(ok) {

+ t.Errorf("validReplication(%q) = false, want true", ok)

+ }

+ for _, bad := range []string{"hourly", "Stream", "off"} {

+ if validReplication(bad) {

+ t.Errorf("validReplication(%q) = true, want false", bad)

+ }

// `add-instance --hibernate` produces a registered, unmaterialized-but-

// spawnable route: everything the activator's spawn needs at first wake (the

// env file with the admin token, a staged binary behind the live symlink, a

diff --git a/internal/activator/replica.go b/internal/activator/replica.go

index b9b8faf..5c92b10 100644

--- a/internal/activator/replica.go

+++ b/internal/activator/replica.go

@@ -200,7 +200,16 @@ func (h *HeldLease) Release(ctx context.Context) error {

// lease's S3 prefix. On success the HeldLease is spent — renewal duty moves to

// the returned LeasedReplicator (Close it to release). On failure the lease is

// STILL HELD; the caller decides whether to retry or Release.

-func (h *HeldLease) Replicate(ctx context.Context, dbPath string) (*LeasedReplicator, error) {

+//

+// syncInterval sets litestream's replica sync tick — the per-plan durability

+// ladder's one knob. It is a MINIMUM spacing between S3 syncs (the monitor's

+// change-notify only short-circuits the idle wait, never this spacing), so 5m

+// bounds an awake host's hard-crash loss to ~5m and 24h to ~a day, while the

+// wake-time initial pass and the sleep-time SyncAndWait/Close flush in full

+// regardless — the interval never weakens hibernation's parked copy. 0 keeps

+// litestream's 1s default (the stream class): idle ticks make zero S3 calls,

+// so stream's request bill is change-driven, not clock-driven.

+func (h *HeldLease) Replicate(ctx context.Context, dbPath string, syncInterval time.Duration) (*LeasedReplicator, error) {

if err := h.rn.Err(); err != nil {

return nil, fmt.Errorf("lease already lost: %w", err)

}

@@ -220,6 +229,9 @@ func (h *HeldLease) Replicate(ctx context.Context, dbPath string) (*LeasedReplic

client.Path = h.leaser.Path

client.Region = h.region

db.Replica = litestream.NewReplicaWithClient(db, client)

+ if syncInterval > 0 {

+ db.Replica.SyncInterval = syncInterval

+ }

store := litestream.NewStore([]*litestream.DB{db}, compactionLevels)

store.L0RetentionCheckInterval = l0RetentionCheckInterval

@@ -246,15 +258,16 @@ type LeasedReplicator struct {

// ReplicateWithLease acquires the single-writer lease at

// s3://bucket/prefix/lock.json and, only if that succeeds, starts litestream

-// replication of dbPath to s3://bucket/prefix. If another process holds the

-// lease it fails LOUDLY with a *litestream.LeaseExistsError (check with

-// errors.As) — this is the signal the activator reacts to.

+// replication of dbPath to s3://bucket/prefix at the default (stream-class)

+// sync interval. If another process holds the lease it fails LOUDLY with a

+// *litestream.LeaseExistsError (check with errors.As) — this is the signal

+// the activator reacts to.

func ReplicateWithLease(ctx context.Context, dbPath, bucket, prefix, region string) (*LeasedReplicator, error) {

held, err := AcquireLease(ctx, bucket, prefix, region)

if err != nil {

return nil, err

}

- r, err := held.Replicate(ctx, dbPath)

+ r, err := held.Replicate(ctx, dbPath, 0)

if err != nil {

_ = held.Release(ctx)

return nil, err

diff --git a/main.go b/main.go

index 0ead4ed..d8a7076 100644

--- a/main.go

+++ b/main.go

@@ -172,6 +172,9 @@ func main() {

case "hibernate":

runHibernate(os.Args[2:])

return

+ case "replication":

+ runReplication(os.Args[2:])

+ return

case "admin-link":

runAdminLink(os.Args[2:])

return

diff --git a/meshreplicate.go b/meshreplicate.go

index ab697af..55ec791 100644

--- a/meshreplicate.go

+++ b/meshreplicate.go

@@ -53,6 +53,7 @@ type meshRoutePayload struct {

Host string

Kind string

Hibernate bool

+ Replication string

AppStoreURL string

HomeOrigin string

PushRelayURL string

@@ -64,7 +65,7 @@ type meshRoutePayload struct {

func routeToPayload(r Route) meshRoutePayload {

return meshRoutePayload{

- Host: r.Host, Kind: r.Kind, Hibernate: r.Hibernate,

+ Host: r.Host, Kind: r.Kind, Hibernate: r.Hibernate, Replication: r.Replication,

AppStoreURL: r.AppStoreURL, HomeOrigin: r.HomeOrigin,

PushRelayURL: r.PushRelayURL, PushRelayToken: r.PushRelayToken,

HomeRegion: r.HomeRegion, Provider: r.Provider, Version: r.Version,

@@ -93,6 +94,7 @@ func (p meshRoutePayload) toLocalRoute() (Route, error) {

HomeRegion: p.HomeRegion,

Provider: p.Provider,

Hibernate: p.Hibernate,

+ Replication: p.Replication,

Version: p.Version,

}, nil

}

diff --git a/meshreplicate_test.go b/meshreplicate_test.go

index f918699..2a3d7cf 100644

--- a/meshreplicate_test.go

+++ b/meshreplicate_test.go

@@ -26,6 +26,7 @@ func TestMeshRoutePayloadTrustBoundary(t *testing.T) {

Host: "space1.11ch.at", Kind: "instance", Socket: "/peer/only/space1.sock",

DBPath: "/peer/only/space1.db", Hibernate: true, Version: "v1",

AppStoreURL: "https://apps.example.com", HomeRegion: "eu",

+ Replication: replicationDaily,

}

payload := routeToPayload(route)

@@ -55,6 +56,12 @@ func TestMeshRoutePayloadTrustBoundary(t *testing.T) {

if local.Host != route.Host || local.Version != route.Version || local.AppStoreURL != route.AppStoreURL || local.HomeRegion != route.HomeRegion || !local.Hibernate {

t.Fatalf("content fields did not survive: %+v", local)

}

+ // The replication class rides the mesh too — a space that fails over to a

+ // peer box must keep its durability rung, not silently revert to stream

+ // (safe for the data, wrong for the plan's economics).

+ if local.Replication != replicationDaily {

+ t.Fatalf("replication class did not survive: %q, want %q", local.Replication, replicationDaily)

+ }

if local.Status != "" || local.NextDue != 0 || local.CreatedAt != 0 {

t.Fatalf("receiver-local lifecycle fields should be zero, got %+v", local)

}

diff --git a/registry.go b/registry.go

index 52582a7..75b6f87 100644

--- a/registry.go

+++ b/registry.go

@@ -2,6 +2,7 @@ package main

import (

"database/sql"

+ "fmt"

"strings"

"time"

)

@@ -42,6 +43,16 @@ type Route struct {

// hibernate while the rest of the box is untouched. Flipped by

// `lchat hibernate <host> --on/--off` via setHibernate.

Hibernate bool

+ // Replication is the host's durability class on the activator's S3 replica

+ // — the per-plan ladder ("stream"/"batched"/"daily"; "" = stream, the

+ // historical behavior). It sets how often litestream's replica syncs local

+ // changes to S3 while the host is awake; every graceful path (sleep,

+ // deploy, shutdown) still flushes everything regardless of class, so the

+ // class only bounds what a HARD crash of the box can lose. Inert for

+ // systemd-managed hosts (no activator replication to tune). Applied at

+ // wake time — a change takes effect on the next wake. Flipped by

+ // `lchat replication <host> <class>` via setReplication.

+ Replication string

// NextDue is the due-time index (#550 Part 4 v2): the earliest pending

// scheduled delivery (unix seconds; 0 = none) recorded by the activator when

// the instance hibernates, so the sweeper can wake it in time to fire. In

@@ -51,6 +62,41 @@ type Route struct {

CreatedAt int64

}

+// The replication classes — the durability ladder a plan buys (#853). Class

+// names describe the mechanism, not the plan, so billing can remap tiers to

+// classes without touching the router. All classes flush in full on every

+// graceful sleep/deploy/shutdown; the sync interval below only bounds the

+// hard-crash loss window (and its S3 Tier-1 request bill) while awake.

+const (

+ replicationStream = "stream" // change-driven ~1s sync — loss window seconds

+ replicationBatched = "batched" // sync at most every 5m — loss window minutes

+ replicationDaily = "daily" // sync at wake + every 24h — loss window up to a day

+// validReplication reports whether s names a replication class ("" counts:

+// it's the unset default, stream).

+func validReplication(s string) bool {

+ switch s {

+ case "", replicationStream, replicationBatched, replicationDaily:

+ return true

+ }

+ return false

+// replicationSyncInterval maps a class to the litestream replica sync

+// interval it buys. 0 means "keep litestream's default" (1s, change-driven —

+// idle ticks make zero S3 calls), which is what stream and the unset ""

+// deliberately share so pre-ladder registries keep today's behavior exactly.

+func replicationSyncInterval(class string) time.Duration {

+ switch class {

+ case replicationBatched:

+ return 5 * time.Minute

+ case replicationDaily:

+ return 24 * time.Hour

+ }

+ return 0

func openRegistry(path string) (*Registry, error) {

d, err := sql.Open("sqlite", path)

if err != nil {

@@ -94,6 +140,7 @@ CREATE TABLE IF NOT EXISTS routes (

home_region TEXT NOT NULL DEFAULT '',

provider TEXT NOT NULL DEFAULT '',

hibernate INTEGER NOT NULL DEFAULT 0,

+ replication TEXT NOT NULL DEFAULT '',

next_due INTEGER NOT NULL DEFAULT 0,

created_at INTEGER NOT NULL

);`); err != nil {

@@ -108,6 +155,7 @@ CREATE TABLE IF NOT EXISTS routes (

`ALTER TABLE routes ADD COLUMN provider TEXT NOT NULL DEFAULT ''`,

`ALTER TABLE routes ADD COLUMN next_due INTEGER NOT NULL DEFAULT 0`,

`ALTER TABLE routes ADD COLUMN hibernate INTEGER NOT NULL DEFAULT 0`,

+ `ALTER TABLE routes ADD COLUMN replication TEXT NOT NULL DEFAULT ''`,

} {

if _, err := r.sql.Exec(stmt); err != nil && !strings.Contains(err.Error(), "duplicate column") {

return err

@@ -121,7 +169,7 @@ func (r *Registry) list() ([]Route, error) {

rows, err := r.sql.Query(`

SELECT host, kind, socket, status, last_error, db_path, version, prev_version,

app_store_url, home_origin, push_relay_url, push_relay_token,

- home_region, provider, hibernate, next_due, created_at

+ home_region, provider, hibernate, replication, next_due, created_at

FROM routes ORDER BY host`)

if err != nil {

return nil, err

@@ -132,7 +180,7 @@ FROM routes ORDER BY host`)

var t Route

if err := rows.Scan(&t.Host, &t.Kind, &t.Socket, &t.Status, &t.LastError,

&t.DBPath, &t.Version, &t.PrevVersion, &t.AppStoreURL, &t.HomeOrigin,

- &t.PushRelayURL, &t.PushRelayToken, &t.HomeRegion, &t.Provider, &t.Hibernate, &t.NextDue, &t.CreatedAt); err != nil {

+ &t.PushRelayURL, &t.PushRelayToken, &t.HomeRegion, &t.Provider, &t.Hibernate, &t.Replication, &t.NextDue, &t.CreatedAt); err != nil {

return nil, err

}

out = append(out, t)

@@ -146,10 +194,10 @@ func (r *Registry) get(host string) (*Route, error) {

err := r.sql.QueryRow(`

SELECT host, kind, socket, status, last_error, db_path, version, prev_version,

app_store_url, home_origin, push_relay_url, push_relay_token,

- home_region, provider, hibernate, next_due, created_at

+ home_region, provider, hibernate, replication, next_due, created_at

FROM routes WHERE host = ?`, host).Scan(&t.Host, &t.Kind, &t.Socket, &t.Status,

&t.LastError, &t.DBPath, &t.Version, &t.PrevVersion, &t.AppStoreURL, &t.HomeOrigin,

- &t.PushRelayURL, &t.PushRelayToken, &t.HomeRegion, &t.Provider, &t.Hibernate, &t.NextDue, &t.CreatedAt)

+ &t.PushRelayURL, &t.PushRelayToken, &t.HomeRegion, &t.Provider, &t.Hibernate, &t.Replication, &t.NextDue, &t.CreatedAt)

if err == sql.ErrNoRows {

return nil, nil

}

@@ -182,8 +230,8 @@ func (r *Registry) upsert(t Route) error {

_, err := r.sql.Exec(`

INSERT INTO routes (host, kind, socket, status, last_error, db_path, version, prev_version,

app_store_url, home_origin, push_relay_url, push_relay_token,

- home_region, provider, hibernate, next_due, created_at)

-VALUES (?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?)

+ home_region, provider, hibernate, replication, next_due, created_at)

+VALUES (?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?)

ON CONFLICT(host) DO UPDATE SET

kind=excluded.kind, socket=excluded.socket, status=excluded.status,

last_error=excluded.last_error, db_path=excluded.db_path, version=excluded.version,

@@ -191,10 +239,10 @@ ON CONFLICT(host) DO UPDATE SET

app_store_url=excluded.app_store_url, home_origin=excluded.home_origin,

push_relay_url=excluded.push_relay_url, push_relay_token=excluded.push_relay_token,

home_region=excluded.home_region, provider=excluded.provider,

- hibernate=excluded.hibernate, next_due=excluded.next_due`,

+ hibernate=excluded.hibernate, replication=excluded.replication, next_due=excluded.next_due`,

t.Host, t.Kind, t.Socket, t.Status, t.LastError, t.DBPath, t.Version, t.PrevVersion,

t.AppStoreURL, t.HomeOrigin, t.PushRelayURL, t.PushRelayToken,

- t.HomeRegion, t.Provider, t.Hibernate, t.NextDue, t.CreatedAt)

+ t.HomeRegion, t.Provider, t.Hibernate, t.Replication, t.NextDue, t.CreatedAt)

return err

}

@@ -207,6 +255,23 @@ func (r *Registry) setHibernate(host string, on bool) error {

return err

}

+// setReplication records host's replication class, same narrow-UPDATE contract

+// as setHibernate. Validated here (not just at the CLI) so no LOCAL write path

+// — operator, controld, or a future billing push — can typo a class the wake

+// path would silently treat as stream. upsert deliberately does NOT validate:

+// a mesh peer on a newer build may replicate a class this build doesn't know,

+// and storing it verbatim (replicationSyncInterval fail-safes unknowns to

+// stream) keeps the space's failover working through the skew, converging when

+// this box updates. An unknown host is a quiet no-op.

+func (r *Registry) setReplication(host, class string) error {

+ if !validReplication(class) {

+ return fmt.Errorf("unknown replication class %q (want %s, %s or %s)",

+ class, replicationStream, replicationBatched, replicationDaily)

+ }

+ _, err := r.sql.Exec(`UPDATE routes SET replication = ? WHERE host = ?`, class, host)

+ return err

// setVersions records a host's current + previous version, same narrow-UPDATE

// contract as setHibernate/setNextDue. Deploys and update passes hold their

// route reads for a while (minutes, when a download runs) — a full upsert