| 1 | package main |
| 2 | |
| 3 | // Billing → replication class (the policy half of the durability ladder; |
| 4 | // registry.go has the mechanism). The cloud provisioner maps each space's |
| 5 | // billing state to a replication class and pushes it to controld, which |
| 6 | // records it on the registry route — the same event-driven-plus-sweep shape |
| 7 | // as storage caps (meter.go): pushed at creation (born classed), on every |
| 8 | // Stripe entitlement move, and reconciled by the meter sweep as the |
| 9 | // self-healing backstop. Classes are config (LCHAT_CLOUD_REPLICATION_CLASSES); |
| 10 | // no config = no pushes anywhere, and operator-run instances outside the |
| 11 | // ledger are never touched. |
| 12 | |
| 13 | import ( |
| 14 | "bytes" |
| 15 | "encoding/json" |
| 16 | "fmt" |
| 17 | "log" |
| 18 | "net/http" |
| 19 | "strings" |
| 20 | ) |
| 21 | |
| 22 | // parseReplicationClasses reads LCHAT_CLOUD_REPLICATION_CLASSES: |
| 23 | // "free=daily,family=daily,community=batched,business=stream" — keys are tier |
| 24 | // keys from the price book, plus "free" for unpaid spaces (the same key |
| 25 | // convention as LCHAT_CLOUD_STORAGE_CAPS). |
| 26 | func parseReplicationClasses(v string) (map[string]string, error) { |
| 27 | if strings.TrimSpace(v) == "" { |
| 28 | return nil, nil |
| 29 | } |
| 30 | classes := map[string]string{} |
| 31 | for _, pair := range strings.Split(v, ",") { |
| 32 | k, val, ok := strings.Cut(strings.TrimSpace(pair), "=") |
| 33 | if !ok || strings.TrimSpace(k) == "" { |
| 34 | return nil, fmt.Errorf("bad entry %q (want key=class)", pair) |
| 35 | } |
| 36 | class := strings.ToLower(strings.TrimSpace(val)) |
| 37 | if class == "" || !validReplication(class) { |
| 38 | return nil, fmt.Errorf("class %q: want %s, %s or %s", val, |
| 39 | replicationStream, replicationBatched, replicationDaily) |
| 40 | } |
| 41 | classes[strings.TrimSpace(k)] = class |
| 42 | } |
| 43 | return classes, nil |
| 44 | } |
| 45 | |
| 46 | // replicationFor maps a space's billing state to its replication class. |
| 47 | // "" = nothing to push (classes unconfigured, or free with no "free" entry — |
| 48 | // the host then keeps whatever it has, which for an unclassed route is |
| 49 | // stream). A PAID tier missing from the map gets stream, never a downgrade: |
| 50 | // a misspelled tier key in config must fail toward durability, not away |
| 51 | // from what the member paid for. |
| 52 | func (s *cloudServer) replicationFor(entitlement, tier string) string { |
| 53 | if len(s.replicationClasses) == 0 { |
| 54 | return "" |
| 55 | } |
| 56 | if entitlement == entDemo { |
| 57 | return "" // demos never replicate — they're never activator-managed (demo.go) |
| 58 | } |
| 59 | if entitlement == entPaid { |
| 60 | if class, ok := s.replicationClasses[tier]; ok { |
| 61 | return class |
| 62 | } |
| 63 | return replicationStream |
| 64 | } |
| 65 | return s.replicationClasses["free"] |
| 66 | } |
| 67 | |
| 68 | // controlSetReplication pushes one instance's replication class to controld. |
| 69 | func (s *cloudServer) controlSetReplication(host, class string) error { |
| 70 | payload, _ := json.Marshal(map[string]any{"class": class}) |
| 71 | req, err := http.NewRequest("POST", "http://control/instances/"+host+"/replication", bytes.NewReader(payload)) |
| 72 | if err != nil { |
| 73 | return err |
| 74 | } |
| 75 | req.Header.Set("Authorization", "Bearer "+s.controlToken) |
| 76 | req.Header.Set("Content-Type", "application/json") |
| 77 | resp, err := s.control.Do(req) |
| 78 | if err != nil { |
| 79 | return err |
| 80 | } |
| 81 | defer resp.Body.Close() |
| 82 | if resp.StatusCode != http.StatusOK { |
| 83 | return fmt.Errorf("control replication %s: HTTP %d", host, resp.StatusCode) |
| 84 | } |
| 85 | return nil |
| 86 | } |
| 87 | |
| 88 | // pushReplication applies a space's tier class to its instance. Best-effort |
| 89 | // like pushQuota: a failed push logs and stands down — the meter sweep |
| 90 | // reconciles it on the next pass. Unlike quota there is no ledger record to |
| 91 | // compare against; controld no-ops (and skips the router reload) when the |
| 92 | // class already matches, so re-pushing every sweep is a cheap localhost POST. |
| 93 | func (s *cloudServer) pushReplication(host, entitlement, tier string) { |
| 94 | class := s.replicationFor(entitlement, tier) |
| 95 | if class == "" { |
| 96 | return |
| 97 | } |
| 98 | if err := s.controlSetReplication(host, class); err != nil { |
| 99 | log.Printf("cloud: replication push %s: %v (sweep will retry)", host, err) |
| 100 | return |
| 101 | } |
| 102 | } |
| 103 | |
| 104 | // pushReplicationByID looks a space's billing state up and applies its class — |
| 105 | // the webhook-side entry point, beside pushQuotaByID. |
| 106 | func (s *cloudServer) pushReplicationByID(id int64) { |
| 107 | if len(s.replicationClasses) == 0 { |
| 108 | return |
| 109 | } |
| 110 | m, err := s.db.spaceBillingByID(id) |
| 111 | if err != nil || m.Host == "" { |
| 112 | return // pending create or gone — the sweep reconciles survivors |
| 113 | } |
| 114 | s.pushReplication(m.Host, m.Entitlement, m.Tier) |
| 115 | } |
| 116 | |
amadan