amadan
server: billing drives the replication class — cloud maps tiers to rungs and pushes via controld (#869)
The policy half of the durability ladder (#866 shipped the mechanism). With
LCHAT_CLOUD_REPLICATION_CLASSES set (free=daily,family=daily,community=batched,
business=stream — tier keys from the price book plus 'free'), the cloud
provisioner keeps every ledger space on its plan's rung:
- born classed: the create request carries the free class, so a fresh space
never spends an awake window on the streaming rung
- webhook pushes: checkout + subscription events re-push on every entitlement
move; a lapse drops the space back to the free rung
- sweep reconcile: the meter sweep re-pushes each pass; controld no-ops (and
skips the router reload) when the class already matches
controld gains POST /instances/{host}/replication: validates the class,
records it on the registry route, reloads the router only on actual change.
Nothing restarts and no sleeping space wakes — the class applies at next wake.
An unknown PAID tier maps to stream deliberately: a config typo must fail
toward durability, never away from what the member paid for. Unset env = no
pushes anywhere (the pre-ladder status quo); operator-run instances outside
the ledger are never touched.
Co-authored-by: Claude Fable 5 <noreply@anthropic.com>
8 files changed,
+472
−8
billing.go+4 −2cloud.go+20 −6cloudreplication.go+112 −0cloudreplication_test.go+146 −0controld.go+63 −0controld_test.go+106 −0docs/cloud-provisioner.md+18 −0meter.go+3 −0
diff --git a/billing.go b/billing.go| index 1e14570..d54ac64 100644 |
| --- a/billing.go |
| +++ b/billing.go |
| @@ -405,7 +405,8 @@ func (s *cloudServer) handleCheckoutCompleted(raw json.RawMessage) error { |
| switch err := s.db.markSpacePaid(id, obj.Customer, obj.Subscription, obj.Metadata["tier"]); err { |
| case nil: |
| log.Printf("cloud: space id %d upgraded (checkout complete, tier %s)", id, obj.Metadata["tier"]) |
| - s.pushQuotaByID(id) // raise the space's storage cap to its new tier |
| + s.pushQuotaByID(id) // raise the space's storage cap to its new tier |
| + s.pushReplicationByID(id) // move it onto its tier's durability rung |
| return nil |
| case sql.ErrNoRows: |
| log.Printf("cloud: webhook checkout for unknown space id %d — ignoring", id) |
| @@ -479,7 +480,8 @@ func (s *cloudServer) handleSubscriptionEvent(evType string, raw json.RawMessage |
| switch err := s.db.setEntitlementTier(id, ent, tier, paidUntil, obj.CancelAtPeriodEnd); err { |
| case nil: |
| log.Printf("cloud: space id %d entitlement → %s (tier %s, subscription %s, cancelling=%v)", id, ent, tier, obj.Status, obj.CancelAtPeriodEnd) |
| - s.pushQuotaByID(id) // entitlement/tier moved — re-apply the matching cap |
| + s.pushQuotaByID(id) // entitlement/tier moved — re-apply the matching cap |
| + s.pushReplicationByID(id) // and the matching durability rung (lapse = back to free's) |
| return nil |
| case sql.ErrNoRows: |
| log.Printf("cloud: webhook %s for unknown space id %d — ignoring", evType, id) |
diff --git a/cloud.go b/cloud.go| index d8c6f20..047e712 100644 |
| --- a/cloud.go |
| +++ b/cloud.go |
| @@ -83,7 +83,10 @@ type cloudServer struct { |
| reapMax int // max reclaims per sweep (blast-radius cap; 0 = unlimited) |
| billing *stripeBilling // optional space upgrades (billing.go); nil = not offered |
| storageCaps map[string]int64 // tier key (+ "free") → storage cap pushed to instances; empty = no enforcement |
| - createBudget int // max creates fleet-wide per trailing 24h (0 = uncapped) |
| + // replicationClasses maps tier keys (+ "free") to durability-ladder classes |
| + // (cloudreplication.go); empty = classes never pushed, every space streams. |
| + replicationClasses map[string]string |
| + createBudget int // max creates fleet-wide per trailing 24h (0 = uncapped) |
| } |
| func (s *cloudServer) handler() http.Handler { |
| @@ -237,9 +240,12 @@ func (s *cloudServer) createSpaceHandler(w http.ResponseWriter, r *http.Request) |
| // controlCreate posts one creation request to the control daemon. Every |
| // provisioned space is born free — the entitlement lands in the instance env |
| // so its UI can offer Upgrade from day one; billing flips re-push via the |
| -// quota path. |
| +// quota path. The replication class rides along so the space is born on the |
| +// free rung of the durability ladder (billing flips re-push via |
| +// pushReplicationByID). |
| func (s *cloudServer) controlCreate(host string) (int, map[string]any, error) { |
| - payload, _ := json.Marshal(controlCreateReq{Host: host, Entitlement: entFree}) |
| + payload, _ := json.Marshal(controlCreateReq{Host: host, Entitlement: entFree, |
| + Replication: s.replicationFor(entFree, "")}) |
| req, err := http.NewRequest("POST", "http://control/instances", bytes.NewReader(payload)) |
| if err != nil { |
| return 0, nil, err |
| @@ -438,6 +444,10 @@ func runCloud(args []string) { |
| if err != nil { |
| die("cloud: LCHAT_CLOUD_STORAGE_CAPS: %v", err) |
| } |
| + replicationClasses, err := parseReplicationClasses(os.Getenv("LCHAT_CLOUD_REPLICATION_CLASSES")) |
| + if err != nil { |
| + die("cloud: LCHAT_CLOUD_REPLICATION_CLASSES: %v", err) |
| + } |
| billing, err := newStripeBilling( |
| os.Getenv("LCHAT_STRIPE_SECRET_KEY"), |
| os.Getenv("LCHAT_STRIPE_WEBHOOK_SECRET"), |
| @@ -461,9 +471,10 @@ func runCloud(args []string) { |
| reapAfter: time.Duration(*reapAfterDays) * 24 * time.Hour, |
| reapDryRun: *reapDryRun, |
| reapMax: *reapMax, |
| - billing: billing, |
| - storageCaps: storageCaps, |
| - createBudget: *createBudget, |
| + billing: billing, |
| + storageCaps: storageCaps, |
| + replicationClasses: replicationClasses, |
| + createBudget: *createBudget, |
| } |
| if *createBudget > 0 { |
| log.Printf("cloud: creation budget on (%d/day fleet-wide)", *createBudget) |
| @@ -471,6 +482,9 @@ func runCloud(args []string) { |
| if len(storageCaps) > 0 { |
| log.Printf("cloud: storage caps on (%d entries)", len(storageCaps)) |
| } |
| + if len(replicationClasses) > 0 { |
| + log.Printf("cloud: replication classes on (%d entries)", len(replicationClasses)) |
| + } |
| if billing != nil { |
| log.Printf("cloud: billing on (%d tiers)", len(billing.tiers)) |
| } |
diff --git a/cloudreplication.go b/cloudreplication.go| new file mode 100644 |
| index 0000000..43a6a7c |
| --- /dev/null |
| +++ b/cloudreplication.go |
| @@ -0,0 +1,112 @@ |
| +package main |
| + |
| +// Billing → replication class (the policy half of the durability ladder; |
| +// registry.go has the mechanism). The cloud provisioner maps each space's |
| +// billing state to a replication class and pushes it to controld, which |
| +// records it on the registry route — the same event-driven-plus-sweep shape |
| +// as storage caps (meter.go): pushed at creation (born classed), on every |
| +// Stripe entitlement move, and reconciled by the meter sweep as the |
| +// self-healing backstop. Classes are config (LCHAT_CLOUD_REPLICATION_CLASSES); |
| +// no config = no pushes anywhere, and operator-run instances outside the |
| +// ledger are never touched. |
| + |
| +import ( |
| + "bytes" |
| + "encoding/json" |
| + "fmt" |
| + "log" |
| + "net/http" |
| + "strings" |
| +) |
| + |
| +// parseReplicationClasses reads LCHAT_CLOUD_REPLICATION_CLASSES: |
| +// "free=daily,family=daily,community=batched,business=stream" — keys are tier |
| +// keys from the price book, plus "free" for unpaid spaces (the same key |
| +// convention as LCHAT_CLOUD_STORAGE_CAPS). |
| +func parseReplicationClasses(v string) (map[string]string, error) { |
| + if strings.TrimSpace(v) == "" { |
| + return nil, nil |
| + } |
| + classes := map[string]string{} |
| + for _, pair := range strings.Split(v, ",") { |
| + k, val, ok := strings.Cut(strings.TrimSpace(pair), "=") |
| + if !ok || strings.TrimSpace(k) == "" { |
| + return nil, fmt.Errorf("bad entry %q (want key=class)", pair) |
| + } |
| + class := strings.ToLower(strings.TrimSpace(val)) |
| + if class == "" || !validReplication(class) { |
| + return nil, fmt.Errorf("class %q: want %s, %s or %s", val, |
| + replicationStream, replicationBatched, replicationDaily) |
| + } |
| + classes[strings.TrimSpace(k)] = class |
| + } |
| + return classes, nil |
| +} |
| + |
| +// replicationFor maps a space's billing state to its replication class. |
| +// "" = nothing to push (classes unconfigured, or free with no "free" entry — |
| +// the host then keeps whatever it has, which for an unclassed route is |
| +// stream). A PAID tier missing from the map gets stream, never a downgrade: |
| +// a misspelled tier key in config must fail toward durability, not away |
| +// from what the member paid for. |
| +func (s *cloudServer) replicationFor(entitlement, tier string) string { |
| + if len(s.replicationClasses) == 0 { |
| + return "" |
| + } |
| + if entitlement == entPaid { |
| + if class, ok := s.replicationClasses[tier]; ok { |
| + return class |
| + } |
| + return replicationStream |
| + } |
| + return s.replicationClasses["free"] |
| +} |
| + |
| +// controlSetReplication pushes one instance's replication class to controld. |
| +func (s *cloudServer) controlSetReplication(host, class string) error { |
| + payload, _ := json.Marshal(map[string]any{"class": class}) |
| + req, err := http.NewRequest("POST", "http://control/instances/"+host+"/replication", bytes.NewReader(payload)) |
| + if err != nil { |
| + return err |
| + } |
| + req.Header.Set("Authorization", "Bearer "+s.controlToken) |
| + req.Header.Set("Content-Type", "application/json") |
| + resp, err := s.control.Do(req) |
| + if err != nil { |
| + return err |
| + } |
| + defer resp.Body.Close() |
| + if resp.StatusCode != http.StatusOK { |
| + return fmt.Errorf("control replication %s: HTTP %d", host, resp.StatusCode) |
| + } |
| + return nil |
| +} |
| + |
| +// pushReplication applies a space's tier class to its instance. Best-effort |
| +// like pushQuota: a failed push logs and stands down — the meter sweep |
| +// reconciles it on the next pass. Unlike quota there is no ledger record to |
| +// compare against; controld no-ops (and skips the router reload) when the |
| +// class already matches, so re-pushing every sweep is a cheap localhost POST. |
| +func (s *cloudServer) pushReplication(host, entitlement, tier string) { |
| + class := s.replicationFor(entitlement, tier) |
| + if class == "" { |
| + return |
| + } |
| + if err := s.controlSetReplication(host, class); err != nil { |
| + log.Printf("cloud: replication push %s: %v (sweep will retry)", host, err) |
| + return |
| + } |
| +} |
| + |
| +// pushReplicationByID looks a space's billing state up and applies its class — |
| +// the webhook-side entry point, beside pushQuotaByID. |
| +func (s *cloudServer) pushReplicationByID(id int64) { |
| + if len(s.replicationClasses) == 0 { |
| + return |
| + } |
| + m, err := s.db.spaceBillingByID(id) |
| + if err != nil || m.Host == "" { |
| + return // pending create or gone — the sweep reconciles survivors |
| + } |
| + s.pushReplication(m.Host, m.Entitlement, m.Tier) |
| +} |
diff --git a/cloudreplication_test.go b/cloudreplication_test.go| new file mode 100644 |
| index 0000000..66218b5 |
| --- /dev/null |
| +++ b/cloudreplication_test.go |
| @@ -0,0 +1,146 @@ |
| +package main |
| + |
| +import ( |
| + "encoding/json" |
| + "net/http" |
| + "testing" |
| +) |
| + |
| +func TestParseReplicationClasses(t *testing.T) { |
| + got, err := parseReplicationClasses("free=daily,family=daily,community=batched,business=stream") |
| + if err != nil { |
| + t.Fatal(err) |
| + } |
| + want := map[string]string{"free": "daily", "family": "daily", "community": "batched", "business": "stream"} |
| + if len(got) != len(want) { |
| + t.Fatalf("parsed %v, want %v", got, want) |
| + } |
| + for k, v := range want { |
| + if got[k] != v { |
| + t.Errorf("classes[%q] = %q, want %q", k, got[k], v) |
| + } |
| + } |
| + |
| + // Unset = feature off, not an error. |
| + if m, err := parseReplicationClasses(""); err != nil || m != nil { |
| + t.Errorf("empty env = (%v, %v), want (nil, nil)", m, err) |
| + } |
| + // A class the router doesn't know must fail at startup, not silently |
| + // stream forever on the box. |
| + if _, err := parseReplicationClasses("free=hourly"); err == nil { |
| + t.Error("unknown class accepted") |
| + } |
| + if _, err := parseReplicationClasses("free"); err == nil { |
| + t.Error("entry without = accepted") |
| + } |
| + // An EMPTY class value is rejected too — "" is valid to validReplication |
| + // (it's the unset registry default) but meaningless as config. |
| + if _, err := parseReplicationClasses("free="); err == nil { |
| + t.Error("empty class value accepted") |
| + } |
| +} |
| + |
| +// replicationFor is the policy: configured keys act, free without a "free" |
| +// entry pushes nothing, and an unknown PAID tier fails toward durability |
| +// (stream) — a config typo must never quietly put a paying space on the |
| +// nightly rung. |
| +func TestReplicationFor(t *testing.T) { |
| + s := &cloudServer{replicationClasses: map[string]string{ |
| + "free": "daily", "family": "daily", "community": "batched", "business": "stream", |
| + }} |
| + for _, tc := range []struct { |
| + ent, tier, want string |
| + }{ |
| + {entFree, "", "daily"}, |
| + {entFree, "family", "daily"}, // stale tier on a lapsed row: entitlement wins |
| + {entPaid, "family", "daily"}, |
| + {entPaid, "community", "batched"}, |
| + {entPaid, "business", "stream"}, |
| + {entPaid, "buisness", "stream"}, // typo'd/unknown paid tier → stream, never a downgrade |
| + } { |
| + if got := s.replicationFor(tc.ent, tc.tier); got != tc.want { |
| + t.Errorf("replicationFor(%q, %q) = %q, want %q", tc.ent, tc.tier, got, tc.want) |
| + } |
| + } |
| + |
| + // Unconfigured = push nothing anywhere. |
| + off := &cloudServer{} |
| + if got := off.replicationFor(entPaid, "business"); got != "" { |
| + t.Errorf("unconfigured replicationFor = %q, want \"\"", got) |
| + } |
| + // Configured but no "free" key: free spaces keep whatever they have. |
| + partial := &cloudServer{replicationClasses: map[string]string{"business": "stream"}} |
| + if got := partial.replicationFor(entFree, ""); got != "" { |
| + t.Errorf("free with no free entry = %q, want \"\"", got) |
| + } |
| +} |
| + |
| +// replicationControl fakes controld's stats + replication endpoints, recording |
| +// class pushes (the stats half keeps meterOnce's sweep loop running). |
| +func replicationControl(t *testing.T, pushed map[string]string) string { |
| + t.Helper() |
| + mux := http.NewServeMux() |
| + mux.HandleFunc("GET /instances/{host}/stats", func(w http.ResponseWriter, r *http.Request) { |
| + writeJSON(w, http.StatusOK, map[string]any{"storage_bytes": int64(1)}) |
| + }) |
| + mux.HandleFunc("POST /instances/{host}/replication", func(w http.ResponseWriter, r *http.Request) { |
| + var req struct { |
| + Class string `json:"class"` |
| + } |
| + json.NewDecoder(r.Body).Decode(&req) |
| + pushed[r.PathValue("host")] = req.Class |
| + writeJSON(w, http.StatusOK, map[string]any{"ok": true, "changed": true}) |
| + }) |
| + return fakeControl(t, mux.ServeHTTP) |
| +} |
| + |
| +// The class enforcement loop, mirroring TestQuotaPushOnUpgradeAndSweep: a |
| +// checkout upgrade pushes the tier's class immediately, the meter sweep |
| +// reconciles everyone, and a lapse drops the space back to the free rung. |
| +func TestReplicationPushOnUpgradeAndSweep(t *testing.T) { |
| + db := openTestCloudDB(t) |
| + paidID := insertSpaceRow(t, db, "paid.11ch.at", 1000) |
| + insertSpaceRow(t, db, "free.11ch.at", 1000) |
| + |
| + pushed := map[string]string{} |
| + sock := replicationControl(t, pushed) |
| + s := &cloudServer{ |
| + db: db, control: controlClient(sock), controlToken: "t", |
| + replicationClasses: map[string]string{"free": "daily", "business": "stream"}, |
| + } |
| + |
| + // The upgrade path (what the checkout webhook calls). |
| + if err := db.markSpacePaid(paidID, "cus", "sub", "business"); err != nil { |
| + t.Fatal(err) |
| + } |
| + s.pushReplicationByID(paidID) |
| + if pushed["paid.11ch.at"] != "stream" { |
| + t.Fatalf("upgrade push = %q, want stream", pushed["paid.11ch.at"]) |
| + } |
| + |
| + // The sweep reconciles both (controld's changed=false makes re-pushes |
| + // harmless, so unlike quota there is no pushed-state comparison here). |
| + s.meterOnce() |
| + if pushed["free.11ch.at"] != "daily" { |
| + t.Fatalf("sweep push = %q, want daily", pushed["free.11ch.at"]) |
| + } |
| + |
| + // A lapse (subscription deleted → free) re-applies the free rung. |
| + if err := db.setEntitlement(paidID, entFree, 0); err != nil { |
| + t.Fatal(err) |
| + } |
| + s.pushReplicationByID(paidID) |
| + if pushed["paid.11ch.at"] != "daily" { |
| + t.Fatalf("lapse push = %q, want back to daily", pushed["paid.11ch.at"]) |
| + } |
| + |
| + // No classes configured → nothing pushes anywhere. |
| + pushed2 := map[string]string{} |
| + sock2 := replicationControl(t, pushed2) |
| + s2 := &cloudServer{db: db, control: controlClient(sock2), controlToken: "t"} |
| + s2.meterOnce() |
| + s2.pushReplicationByID(paidID) |
| + if len(pushed2) != 0 { |
| + t.Errorf("unconfigured classes pushed: %v", pushed2) |
| + } |
| +} |
diff --git a/controld.go b/controld.go| index 9062879..3fe772f 100644 |
| --- a/controld.go |
| +++ b/controld.go |
| @@ -84,6 +84,9 @@ type controlCreateReq struct { |
| PushRelayURL string `json:"push_relay_url,omitempty"` |
| PushRelayToken string `json:"push_relay_token,omitempty"` |
| Entitlement string `json:"entitlement,omitempty"` |
| + // Replication classes the space at birth (registry.go's durability ladder) |
| + // so a free space never spends a single awake window on the streaming rung. |
| + Replication string `json:"replication,omitempty"` |
| } |
| // controlErr is the wire error shape: a stable machine code plus a human |
| @@ -103,10 +106,69 @@ func (c *controld) handler() http.Handler { |
| mux.HandleFunc("DELETE /instances/{host}", c.auth(c.deleteHandler)) |
| mux.HandleFunc("GET /instances/{host}/stats", c.auth(c.statsHandler)) |
| mux.HandleFunc("POST /instances/{host}/quota", c.auth(c.quotaHandler)) |
| + mux.HandleFunc("POST /instances/{host}/replication", c.auth(c.replicationHandler)) |
| mux.HandleFunc("POST /instances/{host}/materialize", c.auth(c.materializeHandler)) |
| return mux |
| } |
| +// replicationHandler records one instance's replication class on its registry |
| +// route (the durability ladder — registry.go) and reloads the router so the |
| +// next wake reads it. The class is wake-time config, so nothing is restarted |
| +// and no sleeping space is woken: an awake instance keeps its current sync |
| +// tick until it next sleeps. Idempotent, and deliberately reload-free when the |
| +// class already matches — the cloud sweep re-pushes every pass, and a router |
| +// reload per space per sweep would be pointless churn. |
| +func (c *controld) replicationHandler(w http.ResponseWriter, r *http.Request) { |
| + host := strings.ToLower(strings.TrimSpace(r.PathValue("host"))) |
| + if !validHost(host) { |
| + writeControlErr(w, http.StatusBadRequest, "invalid_host", "not a valid hostname") |
| + return |
| + } |
| + var req struct { |
| + Class string `json:"class"` |
| + } |
| + if err := json.NewDecoder(io.LimitReader(r.Body, 4<<10)).Decode(&req); err != nil { |
| + writeControlErr(w, http.StatusBadRequest, "bad_request", "invalid JSON body") |
| + return |
| + } |
| + class := strings.ToLower(strings.TrimSpace(req.Class)) |
| + if class == "" || !validReplication(class) { |
| + writeControlErr(w, http.StatusBadRequest, "bad_request", |
| + fmt.Sprintf("class must be %s, %s or %s", replicationStream, replicationBatched, replicationDaily)) |
| + return |
| + } |
| + rt, err := c.reg.get(host) |
| + if err != nil { |
| + log.Printf("controld: replication %s: lookup: %v", host, err) |
| + writeControlErr(w, http.StatusInternalServerError, "lookup_failed", "db error") |
| + return |
| + } |
| + if rt == nil || rt.Kind != "instance" { |
| + writeControlErr(w, http.StatusNotFound, "unknown_host", "no such instance") |
| + return |
| + } |
| + if rt.Replication == class { |
| + writeJSON(w, http.StatusOK, map[string]any{"ok": true, "changed": false}) |
| + return |
| + } |
| + if err := c.reg.setReplication(host, class); err != nil { |
| + log.Printf("controld: replication %s: %v", host, err) |
| + writeControlErr(w, http.StatusInternalServerError, "replication_failed", err.Error()) |
| + return |
| + } |
| + // Best-effort: the registry write is durable either way, and the sweep's |
| + // next push finds changed=false and stops — so a failed reload here must be |
| + // loud (it's the only convergence signal left until something else reloads |
| + // the router). |
| + reloaded := true |
| + if err := systemctl("reload", "lchat-router"); err != nil { |
| + reloaded = false |
| + log.Printf("controld: replication %s → %s: router reload FAILED: %v (class is stored but inert until a reload)", host, class, err) |
| + } |
| + log.Printf("controld: replication %s → %s (reloaded=%v)", host, class, reloaded) |
| + writeJSON(w, http.StatusOK, map[string]any{"ok": true, "changed": true, "reloaded": reloaded}) |
| +} |
| + |
| // materializeHandler stages the running control binary and points host's live |
| // symlink at it, if not already done. This is the privileged half of a route's |
| // first wake on a box that never locally created it (control.go's |
| @@ -368,6 +430,7 @@ func (c *controld) createHandler(w http.ResponseWriter, r *http.Request) { |
| PushRelayURL: req.PushRelayURL, |
| PushRelayToken: req.PushRelayToken, |
| Entitlement: req.Entitlement, |
| + Replication: strings.ToLower(strings.TrimSpace(req.Replication)), |
| // Hibernate-by-default (#550): with LCHAT_CONTROL_HIBERNATE_CREATES=1 |
| // every provisioner-created space is born activator-managed — the |
| // create's public-readiness probe is its one (first) wake, minting the |
diff --git a/controld_test.go b/controld_test.go| index 116ff57..386ae47 100644 |
| --- a/controld_test.go |
| +++ b/controld_test.go |
| @@ -155,6 +155,112 @@ func socketTripwire(t *testing.T, host string) *atomic.Int32 { |
| return &dials |
| } |
| +// The replication endpoint: records the class on the registry route and |
| +// reloads the router exactly once per actual change — a matching class is a |
| +// no-op with no reload (the cloud sweep re-pushes every pass), and garbage |
| +// classes / unknown hosts never touch anything. |
| +func TestControldReplicationEndpoint(t *testing.T) { |
| + reg := controlTestEnv(t) |
| + if err := reg.upsert(Route{Host: "space.11ch.at", Kind: "instance", Socket: "/run/s.sock", Hibernate: true}); err != nil { |
| + t.Fatal(err) |
| + } |
| + if err := reg.upsert(Route{Host: "svc.11ch.at", Kind: "service", Socket: "/run/v.sock"}); err != nil { |
| + t.Fatal(err) |
| + } |
| + srv, token := newTestControld(t, reg, 40) |
| + |
| + var reloads int |
| + systemctl = func(args ...string) error { |
| + if len(args) == 2 && args[0] == "reload" { |
| + reloads++ |
| + } |
| + return nil |
| + } |
| + |
| + post := func(host, class, tok string) *http.Response { |
| + t.Helper() |
| + body, _ := json.Marshal(map[string]string{"class": class}) |
| + req, _ := http.NewRequest("POST", srv.URL+"/instances/"+host+"/replication", bytes.NewReader(body)) |
| + if tok != "" { |
| + req.Header.Set("Authorization", "Bearer "+tok) |
| + } |
| + resp, err := http.DefaultClient.Do(req) |
| + if err != nil { |
| + t.Fatal(err) |
| + } |
| + t.Cleanup(func() { resp.Body.Close() }) |
| + return resp |
| + } |
| + |
| + if resp := post("space.11ch.at", "daily", ""); resp.StatusCode != http.StatusUnauthorized { |
| + t.Errorf("no token: status = %d, want 401", resp.StatusCode) |
| + } |
| + if resp := post("space.11ch.at", "hourly", token); resp.StatusCode != http.StatusBadRequest { |
| + t.Errorf("bad class: status = %d, want 400", resp.StatusCode) |
| + } |
| + if resp := post("ghost.11ch.at", "daily", token); resp.StatusCode != http.StatusNotFound { |
| + t.Errorf("unknown host: status = %d, want 404", resp.StatusCode) |
| + } |
| + if resp := post("svc.11ch.at", "daily", token); resp.StatusCode != http.StatusNotFound { |
| + t.Errorf("service route: status = %d, want 404", resp.StatusCode) |
| + } |
| + if rt, _ := reg.get("space.11ch.at"); rt.Replication != "" || reloads != 0 { |
| + t.Fatalf("rejected requests left marks: class=%q reloads=%d", rt.Replication, reloads) |
| + } |
| + |
| + resp := post("space.11ch.at", "daily", token) |
| + if resp.StatusCode != http.StatusOK { |
| + t.Fatalf("set: status = %d, want 200", resp.StatusCode) |
| + } |
| + if body := decodeBody(t, resp); body["changed"] != true { |
| + t.Errorf("set: changed = %v, want true", body["changed"]) |
| + } |
| + if rt, _ := reg.get("space.11ch.at"); rt.Replication != "daily" { |
| + t.Fatalf("class not recorded: %q", rt.Replication) |
| + } |
| + if reloads != 1 { |
| + t.Fatalf("reloads after set = %d, want 1", reloads) |
| + } |
| + |
| + // Idempotent re-push (the sweep's steady state): no change, no reload. |
| + resp = post("space.11ch.at", "daily", token) |
| + if resp.StatusCode != http.StatusOK { |
| + t.Fatalf("re-push: status = %d, want 200", resp.StatusCode) |
| + } |
| + if body := decodeBody(t, resp); body["changed"] != false { |
| + t.Errorf("re-push: changed = %v, want false", body["changed"]) |
| + } |
| + if reloads != 1 { |
| + t.Fatalf("re-push reloaded the router: %d reloads", reloads) |
| + } |
| +} |
| + |
| +// A create request carrying a replication class produces a route born on that |
| +// rung (what the provisioner sends for a fresh free space). |
| +func TestControldCreateBornClassed(t *testing.T) { |
| + reg := controlTestEnv(t) |
| + srv, token := newTestControld(t, reg, 40) |
| + |
| + body, _ := json.Marshal(map[string]string{"host": "born.11ch.at", "replication": "daily"}) |
| + req, _ := http.NewRequest("POST", srv.URL+"/instances", bytes.NewReader(body)) |
| + req.Header.Set("Authorization", "Bearer "+token) |
| + resp, err := http.DefaultClient.Do(req) |
| + if err != nil { |
| + t.Fatal(err) |
| + } |
| + defer resp.Body.Close() |
| + if resp.StatusCode != http.StatusCreated { |
| + t.Fatalf("create: status = %d, want 201", resp.StatusCode) |
| + } |
| + rt, err := reg.get("born.11ch.at") |
| + if err != nil || rt == nil { |
| + t.Fatalf("route: %+v, %v", rt, err) |
| + } |
| + if rt.Replication != "daily" { |
| + t.Errorf("born class = %q, want daily", rt.Replication) |
| + } |
| +} |
| + |
| func TestControldRequiresToken(t *testing.T) { |
| reg := controlTestEnv(t) |
| srv, _ := newTestControld(t, reg, 40) |
diff --git a/docs/cloud-provisioner.md b/docs/cloud-provisioner.md| index 7f98173..8d67468 100644 |
| --- a/docs/cloud-provisioner.md |
| +++ b/docs/cloud-provisioner.md |
| @@ -359,6 +359,24 @@ it. Unset = no enforcement, and operator-run instances (not in this ledger) |
| are never touched. Caps are sized to the box today (see |
| `tier-economics.md`); they rise when attachments move off-box (#550). |
| +**Replication classes (the durability ladder).** With |
| +`LCHAT_CLOUD_REPLICATION_CLASSES` set |
| +(`free=daily,family=daily,community=batched,business=stream` — same key |
| +convention as storage caps), the cloud maps each ledger space's billing state |
| +to a replication class on the activator's S3 replica and pushes it via |
| +controld (`POST /instances/{host}/replication` → the registry route + a router |
| +reload, only on actual change). Same event-driven-plus-sweep shape as caps: |
| +spaces are **born classed** (the create request carries the free class), |
| +Stripe webhooks re-push on every entitlement move (a lapse drops the space |
| +back to the free rung), and the meter sweep reconciles drift. The class is |
| +wake-time config — nothing is restarted, no sleeping space is woken, and an |
| +awake instance keeps its current sync tick until it next sleeps. An unknown |
| +**paid** tier maps to `stream` on purpose: a config typo must fail toward |
| +durability, never away from what the member paid for. Unset = no pushes, |
| +every space streams (the pre-ladder status quo). What the classes mean — |
| +sync cadence, loss windows, and the graceful-flush guarantee — lives with |
| +the mechanism in `registry.go` and the ops guide. |
| + |
| ## Billing (Stripe) — tiers, quota exemption, no metering |
| Optional space upgrades (`billing.go`): a person who created a space can pay a |
diff --git a/meter.go b/meter.go| index 83a0267..f869f76 100644 |
| --- a/meter.go |
| +++ b/meter.go |
| @@ -106,6 +106,9 @@ func (s *cloudServer) meterOnce() (metered int) { |
| if want := s.capFor(sp.Entitlement, sp.Tier); want > 0 && want != sp.QuotaBytes { |
| s.pushQuota(sp.ID, sp.Host, sp.Entitlement, sp.Tier) |
| } |
| + // Replication class reconcile (cloudreplication.go): no ledger record |
| + // to compare, so push each pass — controld no-ops on a matching class. |
| + s.pushReplication(sp.Host, sp.Entitlement, sp.Tier) |
| } |
| if metered > 0 { |
| log.Printf("cloud: meter: refreshed %d space(s)", metered) |