| 1 | // 🤖 Mesh membership heartbeats (issue #550 Part 4 v2, Phase 3 — "the |
| 2 | // activator mesh"). Boxes learn who's alive from the bucket itself: each live |
| 3 | // box PUTs one object under a shared members prefix on an interval, LISTs the |
| 4 | // prefix to see its peers, and DELETEs its object on drain. S3 was chosen over |
| 5 | // gossip (memberlist) deliberately: the bucket is already the mesh's shared |
| 6 | // substrate — the leases and replicas live there — so membership needs no new |
| 7 | // ports or protocols between boxes, which may not even share a network (the |
| 8 | // current prod box plus a future AWS pool). Membership changes are rare |
| 9 | // events; a ~15s poll is plenty. |
| 10 | // |
| 11 | // The KEY is the record: an object key is base64url(peer URL), so one LIST — |
| 12 | // no per-member GETs — yields the whole membership, and LastModified (the |
| 13 | // bucket's clock, not the writer's) is the liveness timestamp, so writer |
| 14 | // clock skew never enters the staleness judgement. The body duplicates the |
| 15 | // URL as JSON purely for operators poking at the bucket. On a versioned |
| 16 | // bucket the beats churn noncurrent versions; the standard keep-N lifecycle |
| 17 | // prunes them. |
| 18 | |
| 19 | package activator |
| 20 | |
| 21 | import ( |
| 22 | "bytes" |
| 23 | "context" |
| 24 | "encoding/base64" |
| 25 | "encoding/json" |
| 26 | "fmt" |
| 27 | "io" |
| 28 | "path" |
| 29 | "strings" |
| 30 | "time" |
| 31 | |
| 32 | "github.com/aws/aws-sdk-go-v2/aws" |
| 33 | awsconfig "github.com/aws/aws-sdk-go-v2/config" |
| 34 | awss3 "github.com/aws/aws-sdk-go-v2/service/s3" |
| 35 | ) |
| 36 | |
| 37 | // Heartbeat is one box's liveness record in the mesh membership prefix. |
| 38 | type Heartbeat struct { |
| 39 | Peer string // the box's peer base URL — its mesh identity |
| 40 | LastSeen time.Time // the object's S3 LastModified — the bucket's clock |
| 41 | } |
| 42 | |
| 43 | // HeartbeatStore reads and writes box heartbeats under s3://bucket/prefix/. |
| 44 | type HeartbeatStore struct { |
| 45 | client *awss3.Client |
| 46 | bucket string |
| 47 | prefix string |
| 48 | } |
| 49 | |
| 50 | // NewHeartbeatStore builds a store on the default AWS credential chain, like |
| 51 | // AcquireLease. Region may be empty; the chain resolves it. |
| 52 | func NewHeartbeatStore(ctx context.Context, bucket, prefix, region string) (*HeartbeatStore, error) { |
| 53 | var opts []func(*awsconfig.LoadOptions) error |
| 54 | if region != "" { |
| 55 | opts = append(opts, awsconfig.WithRegion(region)) |
| 56 | } |
| 57 | cfg, err := awsconfig.LoadDefaultConfig(ctx, opts...) |
| 58 | if err != nil { |
| 59 | return nil, fmt.Errorf("load aws config: %w", err) |
| 60 | } |
| 61 | return &HeartbeatStore{ |
| 62 | client: awss3.NewFromConfig(cfg), |
| 63 | bucket: bucket, |
| 64 | prefix: strings.Trim(prefix, "/"), |
| 65 | }, nil |
| 66 | } |
| 67 | |
| 68 | // encodeHeartbeatKey / decodeHeartbeatKey turn a peer URL into an S3-safe key |
| 69 | // leaf and back (URLs carry '/' and ':', which would fake folder structure). |
| 70 | func encodeHeartbeatKey(peer string) string { |
| 71 | return base64.RawURLEncoding.EncodeToString([]byte(peer)) |
| 72 | } |
| 73 | |
| 74 | func decodeHeartbeatKey(leaf string) (string, error) { |
| 75 | b, err := base64.RawURLEncoding.DecodeString(leaf) |
| 76 | if err != nil { |
| 77 | return "", err |
| 78 | } |
| 79 | return string(b), nil |
| 80 | } |
| 81 | |
| 82 | func (s *HeartbeatStore) key(peer string) string { |
| 83 | return path.Join(s.prefix, encodeHeartbeatKey(peer)) |
| 84 | } |
| 85 | |
| 86 | // statsKey is a SIBLING object next to the membership heartbeat, under a |
| 87 | // separate "<prefix>-stats/" path — never mixed into the heartbeat body |
| 88 | // itself, so the hot membership path (List, polled every meshHeartbeatInterval |
| 89 | // for HRW placement) never pays to write or parse a body it doesn't need. |
| 90 | // Stats are read far less often (deploy/status-publish.sh's 5-minute Fleet |
| 91 | // publish), by a different reader (bash, via the AWS CLI, not this package). |
| 92 | func (s *HeartbeatStore) statsKey(peer string) string { |
| 93 | return path.Join(s.prefix+"-stats", encodeHeartbeatKey(peer)) |
| 94 | } |
| 95 | |
| 96 | // BoxStats is one box's self-published capacity snapshot: band words and |
| 97 | // counts only, matching the Fleet dashboard's public disclosure rule |
| 98 | // (status.md — "never remaining slot counts... percentages, or cap values"). |
| 99 | // Awake/Asleep are nil when unknown (never guessed, same convention as |
| 100 | // fleet.json's own awake/asleep). Load is a band word (comfortable/busy/ |
| 101 | // critical) or "" when unknown — never a raw memory ratio. |
| 102 | type BoxStats struct { |
| 103 | // Label is this box's own whimsical, hash-derived name ("ada-stargazing", |
| 104 | // meshstats.go's boxLabel — the exact same algorithm and word lists as |
| 105 | // deploy/status-publish.sh's box_label, kept in sync by hand: neither |
| 106 | // side can compute the OTHER box's label, since it hashes a machine id |
| 107 | // only that box has, so each box must self-report the label, never the |
| 108 | // id it was derived from. |
| 109 | Label string `json:"label"` |
| 110 | Awake *int `json:"awake"` |
| 111 | Asleep *int `json:"asleep"` |
| 112 | Load string `json:"load,omitempty"` |
| 113 | // Region is this box's own coarse continent+country string ("EU · |
| 114 | // Finland" — see status.md's PRIVACY rule), set only via the operator's |
| 115 | // explicit LCHAT_BOX_REGION env (meshstats.go's boxRegion) — unlike the |
| 116 | // local box's own status-publish.sh reading, a peer has no cloud-metadata |
| 117 | // auto-detect here, so absent env means omitted, never guessed. |
| 118 | Region string `json:"region,omitempty"` |
| 119 | } |
| 120 | |
| 121 | // PutStats writes this box's stats snapshot. Called on its own slower cadence |
| 122 | // (meshstats.go), independent of the membership beat. |
| 123 | func (s *HeartbeatStore) PutStats(ctx context.Context, peer string, stats BoxStats) error { |
| 124 | body, err := json.Marshal(stats) |
| 125 | if err != nil { |
| 126 | return err |
| 127 | } |
| 128 | _, err = s.client.PutObject(ctx, &awss3.PutObjectInput{ |
| 129 | Bucket: aws.String(s.bucket), |
| 130 | Key: aws.String(s.statsKey(peer)), |
| 131 | Body: bytes.NewReader(body), |
| 132 | }) |
| 133 | return err |
| 134 | } |
| 135 | |
| 136 | // Put writes (or refreshes) this box's heartbeat. The write itself is the |
| 137 | // timestamp — LastModified advances on every beat. The body also carries the |
| 138 | // box's placement region (LCHAT_MESH_REGION), so peers can learn it via Regions |
| 139 | // and scope placement to a region (mesh.go's candidatesFor). region is omitted |
| 140 | // when empty, so a region-agnostic box's body stays byte-identical to before |
| 141 | // regions existed. The KEY is unchanged either way (base64url(peer)), so |
| 142 | // membership/List and every operator tool that decodes the key |
| 143 | // (deploy/*-posture.sh) are untouched — region rides the body, never the key. |
| 144 | func (s *HeartbeatStore) Put(ctx context.Context, peer, region string) error { |
| 145 | rec := map[string]string{"peer": peer} |
| 146 | if region != "" { |
| 147 | rec["region"] = region |
| 148 | } |
| 149 | body, err := json.Marshal(rec) |
| 150 | if err != nil { |
| 151 | return err |
| 152 | } |
| 153 | _, err = s.client.PutObject(ctx, &awss3.PutObjectInput{ |
| 154 | Bucket: aws.String(s.bucket), |
| 155 | Key: aws.String(s.key(peer)), |
| 156 | Body: bytes.NewReader(body), |
| 157 | }) |
| 158 | return err |
| 159 | } |
| 160 | |
| 161 | // List returns every heartbeat under the prefix. Objects whose key leaf isn't |
| 162 | // a decodable peer record are skipped — a foreign object in the prefix must |
| 163 | // never poison membership. |
| 164 | func (s *HeartbeatStore) List(ctx context.Context) ([]Heartbeat, error) { |
| 165 | var out []Heartbeat |
| 166 | p := awss3.NewListObjectsV2Paginator(s.client, &awss3.ListObjectsV2Input{ |
| 167 | Bucket: aws.String(s.bucket), |
| 168 | Prefix: aws.String(s.prefix + "/"), |
| 169 | }) |
| 170 | for p.HasMorePages() { |
| 171 | page, err := p.NextPage(ctx) |
| 172 | if err != nil { |
| 173 | return nil, err |
| 174 | } |
| 175 | for _, obj := range page.Contents { |
| 176 | peer, err := decodeHeartbeatKey(path.Base(aws.ToString(obj.Key))) |
| 177 | if err != nil { |
| 178 | continue |
| 179 | } |
| 180 | var last time.Time |
| 181 | if obj.LastModified != nil { |
| 182 | last = *obj.LastModified |
| 183 | } |
| 184 | out = append(out, Heartbeat{Peer: peer, LastSeen: last}) |
| 185 | } |
| 186 | } |
| 187 | return out, nil |
| 188 | } |
| 189 | |
| 190 | // Regions returns each peer's advertised placement region (the "region" field |
| 191 | // its heartbeat body carries — see Put), keyed by peer URL. Deliberately |
| 192 | // SEPARATE from List: List reads membership + liveness from keys + LastModified |
| 193 | // alone (no per-member GET — the property the whole membership design rests |
| 194 | // on), whereas region lives in the body, so this GETs each small object. The |
| 195 | // mesh calls it off the hot request path, on its beat, into an in-memory cache |
| 196 | // — region changes rarely (only on relocation) and a box only needs peers' |
| 197 | // regions to scope placement. Best-effort per object: a body that won't GET or |
| 198 | // decode, or that advertises no region (an older box, or a region-agnostic |
| 199 | // one), is simply absent from the map, which the caller treats as agnostic. |
| 200 | func (s *HeartbeatStore) Regions(ctx context.Context) (map[string]string, error) { |
| 201 | out := map[string]string{} |
| 202 | p := awss3.NewListObjectsV2Paginator(s.client, &awss3.ListObjectsV2Input{ |
| 203 | Bucket: aws.String(s.bucket), |
| 204 | Prefix: aws.String(s.prefix + "/"), |
| 205 | }) |
| 206 | for p.HasMorePages() { |
| 207 | page, err := p.NextPage(ctx) |
| 208 | if err != nil { |
| 209 | return nil, err |
| 210 | } |
| 211 | for _, obj := range page.Contents { |
| 212 | key := aws.ToString(obj.Key) |
| 213 | peer, err := decodeHeartbeatKey(path.Base(key)) |
| 214 | if err != nil { |
| 215 | continue // a foreign object must never poison the region view |
| 216 | } |
| 217 | region := s.getRegion(ctx, key) |
| 218 | if region == "" { |
| 219 | continue // no region advertised — leave the peer agnostic |
| 220 | } |
| 221 | out[peer] = region |
| 222 | } |
| 223 | } |
| 224 | return out, nil |
| 225 | } |
| 226 | |
| 227 | // getRegion GETs one heartbeat object and pulls its advertised region from the |
| 228 | // body, "" on any error — a soft miss, never fatal to the region refresh (a |
| 229 | // stale region view at worst scopes a space to the whole fleet for one beat). |
| 230 | func (s *HeartbeatStore) getRegion(ctx context.Context, key string) string { |
| 231 | obj, err := s.client.GetObject(ctx, &awss3.GetObjectInput{ |
| 232 | Bucket: aws.String(s.bucket), |
| 233 | Key: aws.String(key), |
| 234 | }) |
| 235 | if err != nil { |
| 236 | return "" |
| 237 | } |
| 238 | defer obj.Body.Close() |
| 239 | var rec struct { |
| 240 | Region string `json:"region"` |
| 241 | } |
| 242 | if err := json.NewDecoder(io.LimitReader(obj.Body, 1<<12)).Decode(&rec); err != nil { |
| 243 | return "" |
| 244 | } |
| 245 | return rec.Region |
| 246 | } |
| 247 | |
| 248 | // Delete removes this box's heartbeat — the FIRST step of a drain, before any |
| 249 | // checkpoint or lease release, so peers take over immediately. |
| 250 | func (s *HeartbeatStore) Delete(ctx context.Context, peer string) error { |
| 251 | _, err := s.client.DeleteObject(ctx, &awss3.DeleteObjectInput{ |
| 252 | Bucket: aws.String(s.bucket), |
| 253 | Key: aws.String(s.key(peer)), |
| 254 | }) |
| 255 | return err |
| 256 | } |
| 257 | |
amadan