| 1 | package dbreads |
| 2 | |
| 3 | import ( |
| 4 | "context" |
| 5 | "crypto/ed25519" |
| 6 | "crypto/rand" |
| 7 | "encoding/hex" |
| 8 | "encoding/json" |
| 9 | "errors" |
| 10 | "fmt" |
| 11 | "net" |
| 12 | "os" |
| 13 | "sync" |
| 14 | "sync/atomic" |
| 15 | "time" |
| 16 | |
| 17 | mrand "math/rand/v2" |
| 18 | |
| 19 | "github.com/carlosframework/platform/internal/dbdoor" |
| 20 | "github.com/carlosframework/platform/internal/hostclaim" |
| 21 | "github.com/carlosframework/platform/internal/provision" |
| 22 | "github.com/carlosframework/platform/internal/registry" |
| 23 | "github.com/carlosframework/platform/internal/release" |
| 24 | ) |
| 25 | |
| 26 | // AgentConfig is what the box's db-reads loop needs (spec §7, §8). |
| 27 | // internal/host fills it from the box's environment; tests fill it with |
| 28 | // stand-ins. |
| 29 | type AgentConfig struct { |
| 30 | Store release.Store // the deployment bucket |
| 31 | Region string // CARLOS_REGION |
| 32 | Box string // this box's label, as its edges file names it |
| 33 | Boot string // this agent process's id |
| 34 | Pub ed25519.PublicKey // CARLOS_RELEASE_PUBKEY |
| 35 | HibernatePrefix string // CARLOS_HIBERNATE_PREFIX |
| 36 | // OpenReplica opens the read side of this box's own region bucket at a |
| 37 | // prefix the agent derived (lease.ReadReplicaClient in production). |
| 38 | OpenReplica func(ctx context.Context, prefix string) (dbdoor.ReplicaReader, error) |
| 39 | // DialWorker connects to carlos-dbread.socket. |
| 40 | DialWorker func(ctx context.Context) (net.Conn, error) |
| 41 | // Units counts carlos-dbread@ units active, activating or deactivating. |
| 42 | Units func(ctx context.Context) (int, error) |
| 43 | // Free reads the filesystem holding /var/tmp. |
| 44 | Free func() (avail, total uint64, err error) |
| 45 | MaxRestored int64 // DefaultMaxRestored when zero |
| 46 | // MaxPull is a pull's cap on its planned LTX files: MaxPullBytes when |
| 47 | // zero or larger, so no box ships more than a CLI accepts, and never |
| 48 | // more than the worker holds (PlanBytesFactor times MaxRestored). |
| 49 | MaxPull int64 |
| 50 | Now func() time.Time |
| 51 | Jitter func() time.Duration // 0 to MaxJitter |
| 52 | Logf func(format string, args ...any) |
| 53 | // Draining is this box's own published draining flag (its edges file's |
| 54 | // draining field); a draining box takes no new read. Never draining |
| 55 | // when nil. |
| 56 | Draining func() bool |
| 57 | |
| 58 | // Test seams; zero is the spec's value. |
| 59 | OwnerDeadline time.Duration |
| 60 | PlanBudget time.Duration |
| 61 | CancelEvery time.Duration |
| 62 | ChunkEvery time.Duration |
| 63 | MaxOutput int64 |
| 64 | } |
| 65 | |
| 66 | // Agent is the box's half of forwarded reads (spec §8.1): it lists the |
| 67 | // region's queue, verifies requests, elects an owner, runs the identity |
| 68 | // bracket around planning and download, streams bytes to a worker, and |
| 69 | // seals and writes output. It never opens SQLite, decodes pages, runs SQL or |
| 70 | // writes a file. |
| 71 | type Agent struct { |
| 72 | cfg AgentConfig |
| 73 | adm *admission |
| 74 | |
| 75 | mu sync.Mutex |
| 76 | decided map[string]time.Time |
| 77 | |
| 78 | probeOK atomic.Bool |
| 79 | probeAfter chan struct{} |
| 80 | reads sync.WaitGroup |
| 81 | } |
| 82 | |
| 83 | func NewAgent(cfg AgentConfig) *Agent { |
| 84 | if cfg.MaxRestored <= 0 { |
| 85 | cfg.MaxRestored = DefaultMaxRestored |
| 86 | } |
| 87 | if cfg.MaxPull <= 0 || cfg.MaxPull > MaxPullBytes { |
| 88 | cfg.MaxPull = MaxPullBytes |
| 89 | } |
| 90 | // The worker refuses a plan whose files total more than this (its |
| 91 | // planWithinBounds), so a pull over it ends too-large here instead. |
| 92 | if held := PlanBytesFactor * cfg.MaxRestored; cfg.MaxPull > held { |
| 93 | cfg.MaxPull = held |
| 94 | } |
| 95 | if cfg.Now == nil { |
| 96 | cfg.Now = time.Now |
| 97 | } |
| 98 | if cfg.Jitter == nil { |
| 99 | cfg.Jitter = func() time.Duration { return mrand.N(MaxJitter) } |
| 100 | } |
| 101 | if cfg.Logf == nil { |
| 102 | cfg.Logf = func(string, ...any) {} |
| 103 | } |
| 104 | if cfg.Draining == nil { |
| 105 | cfg.Draining = func() bool { return false } |
| 106 | } |
| 107 | if cfg.OwnerDeadline == 0 { |
| 108 | cfg.OwnerDeadline = OwnerDeadline |
| 109 | } |
| 110 | if cfg.PlanBudget == 0 { |
| 111 | cfg.PlanBudget = PlanBudget |
| 112 | } |
| 113 | if cfg.CancelEvery == 0 { |
| 114 | cfg.CancelEvery = CancelEvery |
| 115 | } |
| 116 | if cfg.ChunkEvery == 0 { |
| 117 | cfg.ChunkEvery = ChunkEvery |
| 118 | } |
| 119 | if cfg.MaxOutput == 0 { |
| 120 | cfg.MaxOutput = MaxOutputBytes |
| 121 | } |
| 122 | return &Agent{cfg: cfg, adm: newAdmission(cfg.Units, cfg.Free), decided: map[string]time.Time{}, |
| 123 | probeAfter: make(chan struct{}, 1)} |
| 124 | } |
| 125 | |
| 126 | // Capable is whether this box may publish db-read-1: its last readiness |
| 127 | // probe passed (spec §7.1). |
| 128 | func (a *Agent) Capable() bool { return a.probeOK.Load() } |
| 129 | |
| 130 | // Wait blocks until every read this agent owns has written its result. |
| 131 | func (a *Agent) Wait() { a.reads.Wait() } |
| 132 | |
| 133 | func (a *Agent) now() time.Time { return a.cfg.Now().UTC() } |
| 134 | |
| 135 | func (a *Agent) logf(format string, args ...any) { a.cfg.Logf(format, args...) } |
| 136 | |
| 137 | func (a *Agent) decide(id string) { |
| 138 | a.mu.Lock() |
| 139 | defer a.mu.Unlock() |
| 140 | a.decided[id] = a.now() |
| 141 | } |
| 142 | |
| 143 | func (a *Agent) isDecided(id string) bool { |
| 144 | a.mu.Lock() |
| 145 | defer a.mu.Unlock() |
| 146 | _, ok := a.decided[id] |
| 147 | return ok |
| 148 | } |
| 149 | |
| 150 | // prune forgets decisions older than any listing can show again. |
| 151 | func (a *Agent) prune(now time.Time) { |
| 152 | a.mu.Lock() |
| 153 | defer a.mu.Unlock() |
| 154 | for id, at := range a.decided { |
| 155 | if now.Sub(at) > 2*time.Hour { |
| 156 | delete(a.decided, id) |
| 157 | } |
| 158 | } |
| 159 | } |
| 160 | |
| 161 | // askProbe asks the loop for a probe now: a read's worker failed to start |
| 162 | // (spec §7.1). |
| 163 | func (a *Agent) askProbe() { |
| 164 | select { |
| 165 | case a.probeAfter <- struct{}{}: |
| 166 | default: |
| 167 | } |
| 168 | } |
| 169 | |
| 170 | // closeThenProbe gives the failed worker's connection slot back before |
| 171 | // asking for a probe, so admission sees the worker's unit as the orphan it |
| 172 | // now is and holds the probe until the unit has stopped (plan review |
| 173 | // finding 29). The deferred Close that follows is a no-op (countedConn). |
| 174 | func closeThenProbe(a *Agent, conn net.Conn) { |
| 175 | conn.Close() |
| 176 | a.askProbe() |
| 177 | } |
| 178 | |
| 179 | // countedConn gives a worker connection's slot back exactly once. |
| 180 | type countedConn struct { |
| 181 | net.Conn |
| 182 | once sync.Once |
| 183 | adm *admission |
| 184 | } |
| 185 | |
| 186 | func (c *countedConn) Close() error { |
| 187 | err := c.Conn.Close() |
| 188 | c.once.Do(c.adm.closed) |
| 189 | return err |
| 190 | } |
| 191 | |
| 192 | // dial opens a worker connection inside the three the socket allows. |
| 193 | func (a *Agent) dial(ctx context.Context) (net.Conn, error) { |
| 194 | if !a.adm.opened() { |
| 195 | return nil, errBusy |
| 196 | } |
| 197 | c, err := a.cfg.DialWorker(ctx) |
| 198 | if err != nil { |
| 199 | a.adm.closed() |
| 200 | return nil, err |
| 201 | } |
| 202 | return &countedConn{Conn: c, adm: a.adm}, nil |
| 203 | } |
| 204 | |
| 205 | // check is one half of the identity bracket (spec §8.2): the record read |
| 206 | // fresh, and refused unless it is the instance the envelope named, still a |
| 207 | // replicated instance, still in this box's region and not moving. A class, |
| 208 | // pool or fleet change after the console admitted the read is a record |
| 209 | // change. |
| 210 | func (a *Agent) check(ctx context.Context, req Request) (provision.Record, string, error) { |
| 211 | rec, err := provision.GetRecord(ctx, a.cfg.Store, req.Account, req.App, req.Host) |
| 212 | if errors.Is(err, os.ErrNotExist) { |
| 213 | return provision.Record{}, CodeRecordChanged, nil |
| 214 | } |
| 215 | if err != nil { |
| 216 | return provision.Record{}, "", err |
| 217 | } |
| 218 | inc, ierr := hostclaim.Incarnation(rec.Account, rec.App, rec.Host, rec.CreatedAt) |
| 219 | _, fleet := registry.FleetPlacement(rec.Region) |
| 220 | switch { |
| 221 | case ierr != nil, inc != req.Incarnation, rec.Account != req.Account, rec.App != req.App, |
| 222 | rec.Host != req.Host, !provision.ValidHostShape(rec.Host), ClassRefusal(*rec) != "", |
| 223 | rec.Pool != "", fleet, rec.Region != a.cfg.Region: |
| 224 | return *rec, CodeRecordChanged, nil |
| 225 | case rec.Move != nil: |
| 226 | return *rec, CodeMoving, nil |
| 227 | } |
| 228 | return *rec, "", nil |
| 229 | } |
| 230 | |
| 231 | func nonce() string { |
| 232 | var b [16]byte |
| 233 | rand.Read(b[:]) |
| 234 | return hex.EncodeToString(b[:]) |
| 235 | } |
| 236 | |
| 237 | // elect is J3: the first create of picked owns the read. A create that |
| 238 | // answers an error or a conflict is decided by reading picked back: this |
| 239 | // agent's own nonce owns it, anything else, or a read that fails, does not. |
| 240 | func (a *Agent) elect(ctx context.Context, id string) (Picked, bool) { |
| 241 | p := Picked{Box: a.cfg.Box, Boot: a.cfg.Boot, Nonce: nonce(), Deadline: a.now().Add(a.cfg.OwnerDeadline)} |
| 242 | body, err := json.Marshal(p) |
| 243 | if err != nil { |
| 244 | return Picked{}, false |
| 245 | } |
| 246 | if _, err := a.cfg.Store.Put(ctx, PickedKey(id), body, ""); err == nil { |
| 247 | return p, true |
| 248 | } |
| 249 | raw, _, err := a.cfg.Store.Get(ctx, PickedKey(id)) |
| 250 | if err != nil { |
| 251 | return Picked{}, false |
| 252 | } |
| 253 | var got Picked |
| 254 | if json.Unmarshal(raw, &got) != nil || got.Nonce != p.Nonce { |
| 255 | return Picked{}, false |
| 256 | } |
| 257 | return got, true |
| 258 | } |
| 259 | |
| 260 | // consider decides one in-window queue entry (spec §5.5 step 4): read and |
| 261 | // verify the request against its entry, check admission, wait the jitter, |
| 262 | // and enter the election. A refused or forged request is decided, and so |
| 263 | // logged once. A read that fails admission is not decided: a later pass may |
| 264 | // admit it while the window lasts. |
| 265 | func (a *Agent) consider(ctx context.Context, e QueueEntry) { |
| 266 | raw, _, err := a.cfg.Store.Get(ctx, RequestKey(e.ID)) |
| 267 | if err != nil { |
| 268 | if errors.Is(err, os.ErrNotExist) { |
| 269 | a.decide(e.ID) |
| 270 | a.logf("dbreads: %s: a queue entry with no request; ignored", e.ID) |
| 271 | } |
| 272 | return |
| 273 | } |
| 274 | req, err := Verify(a.cfg.Pub, raw, e.ID, a.cfg.Region, a.now()) |
| 275 | if err == nil && !QueueMatches(e, req) { |
| 276 | err = ErrEnvelope |
| 277 | } |
| 278 | if err != nil { |
| 279 | a.decide(e.ID) |
| 280 | a.logf("dbreads: %s: refused the request: %v", e.ID, err) |
| 281 | return |
| 282 | } |
| 283 | // A request whose console has already given up (its request ended |
| 284 | // before it was announced) is never picked (spec §5.5 step 4). |
| 285 | if _, _, err := a.cfg.Store.Get(ctx, CancelKey(e.ID)); err == nil { |
| 286 | a.decide(e.ID) |
| 287 | return |
| 288 | } |
| 289 | // A draining box does not enter the election (spec §7.1, D2). |
| 290 | if a.cfg.Draining() { |
| 291 | return |
| 292 | } |
| 293 | if err := a.adm.admitRead(ctx, e.ID); err != nil { |
| 294 | return |
| 295 | } |
| 296 | if d := a.cfg.Jitter(); d > 0 { |
| 297 | select { |
| 298 | case <-ctx.Done(): |
| 299 | a.adm.abandon(e.ID) |
| 300 | return |
| 301 | case <-time.After(d): |
| 302 | } |
| 303 | } |
| 304 | // It may have started draining during the jitter. |
| 305 | if a.cfg.Draining() { |
| 306 | a.adm.abandon(e.ID) |
| 307 | return |
| 308 | } |
| 309 | p, owned := a.elect(ctx, e.ID) |
| 310 | a.decide(e.ID) |
| 311 | if !owned { |
| 312 | a.adm.abandon(e.ID) |
| 313 | return |
| 314 | } |
| 315 | a.reads.Add(1) |
| 316 | go a.serve(ctx, req, p) |
| 317 | } |
| 318 | |
| 319 | var errSandbox = errors.New("dbreads: the worker can see more than its sandbox allows") |
| 320 | |
| 321 | // probeRetry is how soon a probe that admission skipped is tried again. |
| 322 | var probeRetry = time.Second |
| 323 | |
| 324 | // Run is the db-reads loop: a probe at start, a pass on every nudge (a |
| 325 | // serial move) and every ListEvery, a probe every ProbeEvery and whenever a |
| 326 | // read's worker failed to start (spec §5.5 step 4, §7.1). A probe that |
| 327 | // admission skips (the failed worker's unit is still stopping) stays owed |
| 328 | // and is tried every probeRetry until it runs (plan review finding 29). |
| 329 | func (a *Agent) Run(ctx context.Context, nudge <-chan struct{}) { |
| 330 | var retry <-chan time.Time |
| 331 | tryProbe := func() { |
| 332 | retry = nil |
| 333 | if !a.Probe(ctx) { |
| 334 | retry = time.After(probeRetry) |
| 335 | } |
| 336 | } |
| 337 | tryProbe() |
| 338 | list := time.NewTicker(ListEvery) |
| 339 | defer list.Stop() |
| 340 | probe := time.NewTicker(ProbeEvery) |
| 341 | defer probe.Stop() |
| 342 | for { |
| 343 | select { |
| 344 | case <-ctx.Done(): |
| 345 | a.reads.Wait() |
| 346 | return |
| 347 | case <-nudge: |
| 348 | a.Pass(ctx) |
| 349 | case <-list.C: |
| 350 | a.Pass(ctx) |
| 351 | case <-probe.C: |
| 352 | tryProbe() |
| 353 | case <-a.probeAfter: |
| 354 | tryProbe() |
| 355 | case <-retry: |
| 356 | tryProbe() |
| 357 | } |
| 358 | } |
| 359 | } |
| 360 | |
| 361 | // Pass lists this region's queue for this hour and the last, and considers |
| 362 | // every in-window entry it has not decided. An out-of-window entry is |
| 363 | // skipped on its key alone. A box whose capability is withdrawn takes no |
| 364 | // new read; one it already holds runs on (spec §11.2). |
| 365 | func (a *Agent) Pass(ctx context.Context) { |
| 366 | if !a.Capable() { |
| 367 | return |
| 368 | } |
| 369 | now := a.now() |
| 370 | a.prune(now) |
| 371 | for _, hour := range []time.Time{now.Add(-time.Hour), now} { |
| 372 | keys, err := a.cfg.Store.List(ctx, QueueHourPrefix(a.cfg.Region, hour)) |
| 373 | if err != nil { |
| 374 | a.logf("dbreads: list the %s queue: %v", a.cfg.Region, err) |
| 375 | continue |
| 376 | } |
| 377 | for _, key := range keys { |
| 378 | e, ok := ParseQueueKey(key) |
| 379 | if !ok || e.Region != a.cfg.Region || !InWindow(e.Issued, now) || a.isDecided(e.ID) { |
| 380 | continue |
| 381 | } |
| 382 | a.consider(ctx, e) |
| 383 | } |
| 384 | } |
| 385 | } |
| 386 | |
| 387 | // Probe is the readiness probe (spec §7.1): a hello on its own connection, |
| 388 | // so a busy box is never mistaken for a broken one. While admission is |
| 389 | // blocked (an orphan unit, or no unit listing) it is skipped, the last |
| 390 | // result stands, and it answers false so Run keeps it owed. |
| 391 | func (a *Agent) Probe(ctx context.Context) bool { |
| 392 | if !a.adm.admitProbe(ctx) { |
| 393 | return false |
| 394 | } |
| 395 | defer a.adm.probeDone() |
| 396 | err := a.probe(ctx) |
| 397 | if err != nil { |
| 398 | a.logf("dbreads: readiness probe: %v", err) |
| 399 | } |
| 400 | a.probeOK.Store(err == nil) |
| 401 | return true |
| 402 | } |
| 403 | |
| 404 | func (a *Agent) probe(ctx context.Context) error { |
| 405 | pctx, cancel := context.WithTimeout(ctx, 30*time.Second) |
| 406 | defer cancel() |
| 407 | conn, err := a.dial(pctx) |
| 408 | if err != nil { |
| 409 | return err |
| 410 | } |
| 411 | defer conn.Close() |
| 412 | defer context.AfterFunc(pctx, func() { conn.Close() })() |
| 413 | if err := writeJSONFrame(conn, frameHello, hello{V: ProtocolVersion}); err != nil { |
| 414 | return err |
| 415 | } |
| 416 | typ, p, err := readFrame(conn) |
| 417 | if err != nil { |
| 418 | return err |
| 419 | } |
| 420 | var h HelloReply |
| 421 | if typ != frameHelloReply || json.Unmarshal(p, &h) != nil { |
| 422 | return errProtocol |
| 423 | } |
| 424 | if h.V != ProtocolVersion { |
| 425 | return fmt.Errorf("dbreads: the worker speaks protocol %d, this agent %d", h.V, ProtocolVersion) |
| 426 | } |
| 427 | if len(h.Errors) != 0 || len(h.Interfaces) != 1 || h.Interfaces[0] != "lo" || h.DataEntries != 0 || h.EtcReadable { |
| 428 | return fmt.Errorf("%w: %+v", errSandbox, h) |
| 429 | } |
| 430 | return nil |
| 431 | } |
| 432 | |
amadan