package host import ( "bufio" "bytes" "context" "crypto/ed25519" "log" "os" "slices" "strconv" "strings" "sync/atomic" "github.com/carlosframework/platform/internal/activator" "github.com/carlosframework/platform/internal/dbdoor" "github.com/carlosframework/platform/internal/dbreads" "github.com/carlosframework/platform/internal/hostwrite" "github.com/carlosframework/platform/internal/lease" "github.com/carlosframework/platform/internal/release" ) const ( dbReadSocketUnit = "carlos-dbread.socket" // dbReadSocketPath is under /run/carlos, the tree every tenant unit hides // (carlos-dbread.socket says why). dbReadSocketPath = "/run/carlos/dbread/worker.sock" dbReadMaxBytesEnv = "CARLOS_DB_READ_MAX_BYTES" // dbPullMaxBytesEnv can only lower a pull's cap (spec §5.7, rev 7). dbPullMaxBytesEnv = "CARLOS_DB_PULL_MAX_BYTES" dbReadsSwitchEnv = "CARLOS_DB_READS" ) type dbReadsEnv struct { region string pub ed25519.PublicKey maxRestored int64 maxPull int64 } // dbReadsSettings is the environment half of spec §7.1's conditions: // hibernation configured, CARLOS_REGION a region, CARLOS_RELEASE_PUBKEY // set, CARLOS_DB_READS not 0, CARLOS_HOST_CLAIM not set. The other half, // a running loop whose last probe passed, is the agent's. why is "" when // the loop may be armed. func dbReadsSettings(getenv func(string) string) (dbReadsEnv, string) { switch { case !hibernationConfigured(getenv): return dbReadsEnv{}, "hibernation is not configured (CARLOS_HIBERNATE_BUCKET)" case hostwrite.Enabled(getenv): return dbReadsEnv{}, "CARLOS_HOST_CLAIM is set" case strings.TrimSpace(getenv(dbReadsSwitchEnv)) == "0": return dbReadsEnv{}, dbReadsSwitchEnv + " is 0" } region := strings.TrimSpace(getenv("CARLOS_REGION")) if !dbreads.ValidRegion(region) { return dbReadsEnv{}, "CARLOS_REGION is not a region" } pub, err := release.ParsePublicKey(getenv("CARLOS_RELEASE_PUBKEY"), "CARLOS_RELEASE_PUBKEY") if err != nil { return dbReadsEnv{}, err.Error() } env := dbReadsEnv{region: region, pub: pub, maxRestored: dbreads.DefaultMaxRestored, maxPull: dbreads.MaxPullBytes} if raw := strings.TrimSpace(getenv(dbReadMaxBytesEnv)); raw != "" { n, err := strconv.ParseInt(raw, 10, 64) if err != nil || n <= 0 { return dbReadsEnv{}, dbReadMaxBytesEnv + " is not a whole number of bytes" } env.maxRestored = n } if raw := strings.TrimSpace(getenv(dbPullMaxBytesEnv)); raw != "" { n, err := strconv.ParseInt(raw, 10, 64) if err != nil || n <= 0 { return dbReadsEnv{}, dbPullMaxBytesEnv + " is not a whole number of bytes" } // Lowering only: no box ships more than a CLI accepts (Review // Focus 5). NewAgent clamps again. env.maxPull = min(n, dbreads.MaxPullBytes) } return env, "" } // armedDBReads is the armed loop; nil on a box that cannot serve. var armedDBReads atomic.Pointer[dbreads.Agent] // dbReadsNudge is the serial pass's word to the loop: list the queue now. var dbReadsNudge = make(chan struct{}, 1) // dbReadsDraining is the draining flag this box last published in its edges // file, which the edges loop stores on every pass. The agent reads the same // value the console's evidence rule reads, so the two never disagree (spec // §7.1). Nothing on main sets an edges file's Draining yet (region HA's // placer will); until then it is false. var dbReadsDraining atomic.Bool func nudgeDBReads() { select { case dbReadsNudge <- struct{}{}: default: } } // dbReadsCapable is whether this box publishes db-read-1: the loop is armed // and its last readiness probe passed (spec §7.1). func dbReadsCapable() bool { a := armedDBReads.Load() return a != nil && a.Capable() } // withDBReadCapability is caps with db-read-1 when capable; caps is not // modified. func withDBReadCapability(caps []string, capable bool) []string { if !capable { return caps } return append(slices.Clone(caps), dbreads.Capability) } // countDBReadUnits reads `systemctl list-units --plain --no-legend` output: // carlos-dbread@ units that are active, activating or deactivating. A // stopped unit's scratch is gone; any other state may still hold it. func countDBReadUnits(out string) int { n := 0 for _, line := range strings.Split(out, "\n") { f := strings.Fields(line) if len(f) < 3 || !strings.HasPrefix(f[0], "carlos-dbread@") { continue } switch f[2] { case "active", "activating", "deactivating", "reloading", "refreshing": n++ } } return n } // armDBReads arms the db-reads loop on a box that can serve (spec §7.1). // Thin, like armRunsPass: the behaviour is dbreads.Agent's, tested there. func armDBReads(hib activator.Config) { env, why := dbReadsSettings(os.Getenv) if why != "" { log.Printf("agent: db reads not armed: %s", why) return } ctx := context.Background() store, err := openDeployStore(ctx) if err != nil { log.Printf("agent: db reads not armed: %v", err) return } a := dbreads.NewAgent(dbreads.AgentConfig{ Store: store, Region: env.region, Box: BoxLabel(), Boot: newBootID(), Pub: env.pub, HibernatePrefix: hib.Prefix, OpenReplica: func(ctx context.Context, prefix string) (dbdoor.ReplicaReader, error) { return lease.ReadReplicaClient(ctx, hib.Bucket, prefix, hib.Region) }, DialWorker: dialDBReadWorker, Units: dbReadUnits, Free: varTmpFree, MaxRestored: env.maxRestored, MaxPull: env.maxPull, Logf: log.Printf, Draining: dbReadsDraining.Load, }) armedDBReads.Store(a) go a.Run(ctx, dbReadsNudge) log.Printf("agent: db reads armed (region %s)", env.region) } // dbReadsHostEnvKeys are the host.env settings dbReadsSettings reads. var dbReadsHostEnvKeys = []string{"CARLOS_HIBERNATE_BUCKET", "CARLOS_REGION", "CARLOS_RELEASE_PUBKEY", dbReadsSwitchEnv, "CARLOS_HOST_CLAIM", dbReadMaxBytesEnv, dbPullMaxBytesEnv} // dbReadsSocketWanted is root's half of the dark gate (Astra round 1, // finding 1): the adopt pass arms carlos-dbread.socket only when the box's // installed host.env, the file the agent's own environment comes from, // arms db reads by the same rule the capability follows (dbReadsSettings). // A host.env it cannot read or parse keeps the socket dark. why is the // reason when it is not wanted. func dbReadsSocketWanted(hostEnv string) (bool, string) { raw, err := readClaimUnitFile(hostEnv) if err != nil { return false, "host.env: " + err.Error() } values := map[string]string{} for _, k := range dbReadsHostEnvKeys { values[k] = "" } scan := bufio.NewScanner(bytes.NewReader(raw)) for scan.Scan() { line := strings.TrimSpace(scan.Text()) if line == "" || strings.HasPrefix(line, "#") || strings.HasPrefix(line, ";") { continue } key, v, ok := strings.Cut(line, "=") key, v = strings.TrimSpace(key), strings.TrimSpace(v) if _, known := values[key]; !ok || !known { continue } if len(v) >= 2 && ((v[0] == '"' && v[len(v)-1] == '"') || (v[0] == '\'' && v[len(v)-1] == '\'')) { v = v[1 : len(v)-1] } if strings.ContainsAny(v, "\\\"'\x00\r\n") { return false, "host.env: unsupported value for " + key } values[key] = v } if err := scan.Err(); err != nil { return false, "host.env: " + err.Error() } if _, why := dbReadsSettings(func(k string) string { return values[k] }); why != "" { return false, why } return true, "" }