| 1 | package console |
| 2 | |
| 3 | import ( |
| 4 | "context" |
| 5 | "errors" |
| 6 | "fmt" |
| 7 | "log" |
| 8 | "os" |
| 9 | "time" |
| 10 | |
| 11 | "github.com/carlosframework/platform/internal/hostactors" |
| 12 | "github.com/carlosframework/platform/internal/hostclaim" |
| 13 | "github.com/carlosframework/platform/internal/hostfence" |
| 14 | "github.com/carlosframework/platform/internal/hostops" |
| 15 | "github.com/carlosframework/platform/internal/provision" |
| 16 | "github.com/carlosframework/platform/internal/release" |
| 17 | ) |
| 18 | |
| 19 | // ConfigureClaimRuntime builds the enabled runtime exactly once, after |
| 20 | // NewServer and before any Start* goroutine or handler can run. It replaces |
| 21 | // every raw handle in s.cfg with its boundary; the raw bucket survives only |
| 22 | // inside the ControlWriter and the boundary's private fields. |
| 23 | func (s *Server) ConfigureClaimRuntime(ctx context.Context) error { |
| 24 | if !s.cfg.HostClaim { |
| 25 | return nil |
| 26 | } |
| 27 | if s.cfg.Releases == nil || s.cfg.DBVersions == nil || s.cfg.Bucket == "" || s.cfg.HostConsoleID == "" || len(s.cfg.Key) == 0 || len(s.cfg.ReleasePub) == 0 { |
| 28 | return hostops.ErrContract |
| 29 | } |
| 30 | if s.claimRuntimeConfigured { |
| 31 | return hostops.ErrContract |
| 32 | } |
| 33 | raw := s.cfg.Releases |
| 34 | control, err := hostops.NewControlWriter(raw, s.cfg.ReleasePub, time.Now, s.cfg.dbClaimHomeRegion()) |
| 35 | if err != nil { |
| 36 | return err |
| 37 | } |
| 38 | check, err := hostops.NewProductionMutationCheck(raw, control) |
| 39 | if err != nil { |
| 40 | return err |
| 41 | } |
| 42 | records, err := hostops.NewConditionalRecordStore(raw, s.cfg.ReleasePub, s.cfg.dbClaimHomeRegion()) |
| 43 | if err != nil { |
| 44 | return err |
| 45 | } |
| 46 | data, err := hostops.NewObservedMutationBoundary(true, records, check, s.claimMutationObserver) |
| 47 | if err != nil { |
| 48 | return err |
| 49 | } |
| 50 | var dataBuckets []string |
| 51 | if s.cfg.HibernateBucket != "" && s.cfg.HibernateBucket != s.cfg.Bucket { |
| 52 | dataBuckets = append(dataBuckets, s.cfg.HibernateBucket) |
| 53 | } |
| 54 | versions, err := hostops.NewVersionedBoundary(true, s.cfg.DBVersions, raw, control, s.cfg.Bucket, dataBuckets, s.claimMutationObserver) |
| 55 | if err != nil { |
| 56 | return err |
| 57 | } |
| 58 | guard := backendGuard{data: data, control: control, observe: s.claimBackendObserver, deployment: s.cfg.StoreDeployment} |
| 59 | |
| 60 | d := hostactors.Directory{Store: data, PublicKey: s.cfg.ReleasePub, Bucket: s.cfg.Bucket} |
| 61 | roster, err := d.Read(ctx) |
| 62 | if err != nil { |
| 63 | return err |
| 64 | } |
| 65 | if !roster.Roster.MatchesHome(s.cfg.dbClaimHomeRegion()) { |
| 66 | return hostops.ErrContract |
| 67 | } |
| 68 | self, ok := roster.Roster.Consoles[s.cfg.HostConsoleID] |
| 69 | if !ok || len(self.Namespaces) != 1 { |
| 70 | return hostops.ErrContract |
| 71 | } |
| 72 | if err = s.store.migrateClaimAppIdentity(); err != nil { |
| 73 | return err |
| 74 | } |
| 75 | a := &hostops.ProductionAuthority{HomeRegion: s.cfg.dbClaimHomeRegion(), Store: data, Control: control, Directory: d, Now: time.Now, ConsoleID: s.cfg.HostConsoleID, Enrollment: self.Enrollment, Namespace: self.Namespaces[0], Signer: release.RootSigner(s.cfg.Key)} |
| 76 | a.Backfill = &hostops.Backfill{Store: data, Directory: d} |
| 77 | // Region HA §5.2 (L8): create's executor from the chooser, only with |
| 78 | // Config.RegionHA on; nil keeps today's create path. |
| 79 | a.Executors = s.executorChooser(d) |
| 80 | a.Evidence = hostactors.ExactEvidence{PublicKey: s.cfg.ReleasePub, Store: data, Boxes: d.PrivateBoxReport, Self: s.localHostReport} |
| 81 | c := hostops.Controller{Store: data, Control: control, Authority: a, Now: time.Now, NewID: claimOperationID} |
| 82 | owner := hostops.RegistrationOwner{Console: s.cfg.HostConsoleID, Enrollment: self.Enrollment, Controller: claimOperationID()} |
| 83 | if owner.Controller == "" { |
| 84 | return hostops.ErrContract |
| 85 | } |
| 86 | |
| 87 | if err = hostops.BindMutationRunner(data, control, owner, s.mutationAppID); err != nil { |
| 88 | return err |
| 89 | } |
| 90 | if err = s.store.bindClaimSQLMutations(data); err != nil { |
| 91 | return err |
| 92 | } |
| 93 | // From here on no field of s.cfg refers to a raw handle. |
| 94 | s.cfg.Releases = data |
| 95 | if s.cfg.WakeRetryReset != nil { |
| 96 | door := *s.cfg.WakeRetryReset |
| 97 | door.Claims = data |
| 98 | if door.production { |
| 99 | door = *productionWakeRetryReset(a) |
| 100 | } |
| 101 | s.cfg.WakeRetryReset = &door |
| 102 | } |
| 103 | s.rebindGeoipStore() |
| 104 | s.cfg.DBVersions = versions |
| 105 | s.cfg.StoreProvisioner = newClaimStoreProvisioner(s.cfg.StoreProvisioner, guard) |
| 106 | s.cfg.StoreScanner = newClaimStoreScanner(s.cfg.StoreScanner, guard) |
| 107 | s.cfg.MailProvisioner = newClaimMailProvisioner(s.cfg.MailProvisioner, guard) |
| 108 | s.cfg.DNSWriter = newClaimDNSWriter(s.cfg.DNSWriter, guard) |
| 109 | |
| 110 | purge := &ClaimPurgeCoordinator{Server: s, Controller: c, Authority: a, Versions: versions, ControlBucket: s.cfg.Bucket, Owner: owner} |
| 111 | placement := &ClaimPlacementCoordinator{Server: s, Controller: c, Authority: a, Owner: owner} |
| 112 | s.cfg.ClaimAuthority = a |
| 113 | s.cfg.HostClaimOps = &c |
| 114 | s.cfg.ClaimPurge = purge |
| 115 | s.cfg.ClaimPlacement = placement |
| 116 | s.claimOwner = owner |
| 117 | s.claimRuntimeConfigured = true |
| 118 | if s.cfg.MovesEnabled && s.cfg.ClaimMoveRecovery == nil { |
| 119 | // The console's move pass (moves Plan 5 Task 7): the moves body of |
| 120 | // claim Plan 10's Recovery.Moves hook, bound only with the moves |
| 121 | // switch on, and only now that every box close is record first |
| 122 | // (ruling M4, P3-6; TestClaimMoveRecoveryWiringWaitsForRecordFirstOwners). |
| 123 | s.cfg.ClaimMoveRecovery = s.movesRecovery |
| 124 | |
| 125 | } |
| 126 | var moveLogs func(context.Context, *hostops.MoveLogWriter) error |
| 127 | if s.cfg.MovesEnabled { |
| 128 | // The delivery of the move pass's log entries (Plan 6 Task 3), once |
| 129 | // a pass and independent of any move's state or grant. |
| 130 | moveLogs = s.drainMoveLog |
| 131 | } |
| 132 | s.cfg.HostClaimRecovery = &hostops.Recovery{ |
| 133 | ScheduleLossPages: s.scheduleClaimLossPages, |
| 134 | ReportAutomaticLosses: func(ctx context.Context, host string) error { |
| 135 | return a.Control.ReportAutomaticLosses(ctx, a.Signer, host) |
| 136 | }, |
| 137 | Moves: s.cfg.ClaimMoveRecovery, |
| 138 | MoveLogs: moveLogs, |
| 139 | MoveFenceAuthority: hostfence.ClaimFenceAuthority{Directory: a.Directory, Now: c.Now}, |
| 140 | Controller: c, |
| 141 | SweepRecords: func(ctx context.Context) error { return c.SweepAllRecords(ctx) }, |
| 142 | SweepPurges: func(ctx context.Context) error { |
| 143 | scanErr := hostops.SweepAllPurges(ctx, c.Store, func(ctx context.Context, app hostops.App, p hostops.Purge) error { |
| 144 | adm, err := hostops.ReadCanonicalAdmission(ctx, c.Store, app.Account, app.Name) |
| 145 | if err != nil { |
| 146 | return err |
| 147 | } |
| 148 | j, _, err := hostops.ReadPurgeJournal(ctx, c.Store, app.Account, app.Name, p.ID, adm.Value.AppID) |
| 149 | if err != nil { |
| 150 | return err |
| 151 | } |
| 152 | return purge.Resume(ctx, app.Account, app.Name, j.ID, j.AppID) |
| 153 | }) |
| 154 | return errors.Join(scanErr, s.SweepClaimPurgeRequests(ctx)) |
| 155 | }, |
| 156 | Purge: func(ctx context.Context, obs hostclaim.Observed) error { |
| 157 | if obs.Claim == nil { |
| 158 | return hostops.ErrIdentity |
| 159 | } |
| 160 | if obs.Claim.Holder == nil { |
| 161 | return hostops.ErrPending |
| 162 | } |
| 163 | h := obs.Claim.Holder |
| 164 | adm, err := hostops.ReadCanonicalAdmission(ctx, c.Store, h.Account, h.App) |
| 165 | if err != nil { |
| 166 | return err |
| 167 | } |
| 168 | j, _, err := hostops.ReadPurgeJournal(ctx, c.Store, h.Account, h.App, obs.Claim.Op.ID, adm.Value.AppID) |
| 169 | if err != nil { |
| 170 | return err |
| 171 | } |
| 172 | return purge.Resume(ctx, h.Account, h.App, j.ID, j.AppID) |
| 173 | }, |
| 174 | Placement: func(ctx context.Context, obs hostclaim.Observed) error { |
| 175 | if obs.Claim == nil || obs.Claim.Holder == nil { |
| 176 | return hostops.ErrIdentity |
| 177 | } |
| 178 | h := obs.Claim.Holder |
| 179 | return placement.Resume(ctx, h.Account, h.App, obs.Claim.Op.Intent["app_id"], obs.Claim.Op.ID) |
| 180 | }, |
| 181 | SweepAdmissions: s.SweepClaimAdmissions, |
| 182 | } |
| 183 | return nil |
| 184 | } |
| 185 | |
| 186 | func (s *Server) SweepClaimAdmissions(ctx context.Context) error { |
| 187 | if !s.cfg.HostClaim { |
| 188 | return nil |
| 189 | } |
| 190 | c := s.cfg.HostClaimOps |
| 191 | if c == nil || c.Control == nil || s.cfg.ClaimPlacement == nil || s.cfg.ClaimPurge == nil { |
| 192 | return hostops.ErrContract |
| 193 | } |
| 194 | // A projection error on one console row must not stop independent |
| 195 | // canonical operations on every other app. |
| 196 | identityErr := s.SweepClaimAppIdentities(ctx) |
| 197 | apps, err := hostops.Apps(ctx, c.Store) |
| 198 | if err != nil { |
| 199 | return errors.Join(identityErr, err) |
| 200 | } |
| 201 | all := []error{identityErr} |
| 202 | for _, app := range apps { |
| 203 | if err := s.sweepOneClaimAdmission(ctx, app); err != nil && !errors.Is(err, hostops.ErrPending) { |
| 204 | all = append(all, err) |
| 205 | } |
| 206 | } |
| 207 | return errors.Join(all...) |
| 208 | } |
| 209 | |
| 210 | func (s *Server) sweepOneClaimAdmission(ctx context.Context, app hostops.App) (result error) { |
| 211 | c := s.cfg.HostClaimOps |
| 212 | a, err := hostops.ReadCanonicalAdmission(ctx, c.Store, app.Account, app.Name) |
| 213 | if errors.Is(err, os.ErrNotExist) { |
| 214 | return nil |
| 215 | } |
| 216 | if err != nil { |
| 217 | return err |
| 218 | } |
| 219 | appID := a.Value.AppID |
| 220 | // SQL errors remain visible but do not prevent unrelated registration or |
| 221 | // operation recovery. Parent operations already refuse unsettled children. |
| 222 | var sqlErrors []error |
| 223 | defer func() { result = errors.Join(result, errors.Join(sqlErrors...)) }() |
| 224 | var database string |
| 225 | var databaseErr error |
| 226 | databaseChecked := false |
| 227 | for id, e := range a.Value.Effects { |
| 228 | if e.Action != "sql-ingest" || e.State == "complete" { |
| 229 | continue |
| 230 | } |
| 231 | after, parseErr := time.Parse(time.RFC3339Nano, e.Inputs["sql_recover_after"]) |
| 232 | reg, registered := a.Value.Registrations[e.OperationID] |
| 233 | if parseErr != nil { |
| 234 | if !registered || hostops.RegistrationLive(reg, c.Now()) { |
| 235 | continue |
| 236 | } |
| 237 | } else if c.Now().Before(after) { |
| 238 | continue |
| 239 | } |
| 240 | // Nothing above this point needs the local receipt database. In |
| 241 | // particular, apps with no outstanding SQL never try to open it. |
| 242 | if !databaseChecked { |
| 243 | databaseChecked = true |
| 244 | database, databaseErr = s.store.sqlDatabaseID() |
| 245 | if databaseErr != nil { |
| 246 | sqlErrors = append(sqlErrors, databaseErr) |
| 247 | } |
| 248 | } |
| 249 | // An identity error here means the file was replaced under this |
| 250 | // running console (review m-4): no effect can settle against it, so |
| 251 | // each is alerted exactly as one bound to a restored-away database. |
| 252 | replacedUnderUs := errors.Is(databaseErr, hostops.ErrIdentity) |
| 253 | if databaseErr != nil && !replacedUnderUs { |
| 254 | continue |
| 255 | } |
| 256 | if replacedUnderUs || e.Inputs["sql_database"] != database { |
| 257 | // Bound to a receipt database this console no longer has (a |
| 258 | // restore or replacement changed its identity). Never inferred: |
| 259 | // it stays unsettled, visible and alerted until an operator |
| 260 | // takes the restore runbook's break-glass path. |
| 261 | sqlErrors = append(sqlErrors, fmt.Errorf("%w: %s/%s effect %s", ErrSQLDatabaseReplaced, app.Account, app.Name, id)) |
| 262 | s.alertSQLDatabaseReplaced(ctx, app.Account, app.Name, id) |
| 263 | continue |
| 264 | } |
| 265 | if err := c.Control.RecoverSQLTransaction(ctx, app.Account, app.Name, appID, id, s.store); err != nil { |
| 266 | sqlErrors = append(sqlErrors, err) |
| 267 | } |
| 268 | } |
| 269 | for id, reg := range a.Value.Registrations { |
| 270 | if reg.Kind != hostops.RegistrationIngest { |
| 271 | continue |
| 272 | } |
| 273 | if err := c.Control.SettleRegistration(ctx, app.Account, app.Name, appID, id, c.Now()); err != nil && !errors.Is(err, hostops.ErrPending) { |
| 274 | return err |
| 275 | } |
| 276 | } |
| 277 | if op := a.Value.Op; op != nil { |
| 278 | switch op.Kind { |
| 279 | case hostops.OpAppMutation: |
| 280 | if hostops.IsRestoreOp(op) { |
| 281 | // Review P2-4: a receipt-bound restore is finished from the |
| 282 | // receipt (prepared receipt with no effect included); a |
| 283 | // receipt-less one is AbandonRestore's (sql-work), which |
| 284 | // never settles an issued or uncertain effect. |
| 285 | return s.reconcileClaimRestore(ctx, app.Account, app.Name, appID) |
| 286 | } |
| 287 | // Mail members reopen only after their gate is released or |
| 288 | // reconciled (RecoverMutation's mail branch). |
| 289 | return c.Control.RecoverMutation(ctx, app.Account, app.Name, appID, op.ID) |
| 290 | case hostops.OpPlacing: |
| 291 | return s.cfg.ClaimPlacement.Resume(ctx, app.Account, app.Name, appID, op.ID) |
| 292 | case hostops.OpPurging: |
| 293 | return s.cfg.ClaimPurge.Resume(ctx, app.Account, app.Name, op.ID, appID) |
| 294 | case hostops.OpDeleting: |
| 295 | // Resume auxiliary cleanup even after the host claim became free. |
| 296 | return s.retireClaimHostRecord(ctx, provision.RetireRef{Account: app.Account, App: app.Name, Host: op.Intent["host"], Family: op.Intent["family"], Name: op.Intent["name"], Incarnation: op.Intent["incarnation"], By: "console-recovery"}) |
| 297 | default: |
| 298 | return hostops.ErrContract |
| 299 | } |
| 300 | } |
| 301 | var all []error |
| 302 | for id := range a.Value.Registrations { |
| 303 | if err := c.Control.SettleRegistration(ctx, app.Account, app.Name, appID, id, c.Now()); err != nil && !errors.Is(err, hostops.ErrPending) { |
| 304 | all = append(all, err) |
| 305 | } |
| 306 | } |
| 307 | return errors.Join(all...) |
| 308 | } |
| 309 | |
| 310 | // ErrSQLDatabaseReplaced is an SQL effect past sql_recover_after whose |
| 311 | // receipt database is not this console's current one. |
| 312 | var ErrSQLDatabaseReplaced = errors.New("console: SQL effect bound to a replaced receipt database") |
| 313 | |
| 314 | // sqlReceiptAlertBox is the alert_seen box for receipt database alerts. |
| 315 | const sqlReceiptAlertBox = "claim-sql" |
| 316 | |
| 317 | func sqlReplacedRule(account, app, effect string) string { |
| 318 | return "replaced-database " + account + "/" + app + "/" + effect |
| 319 | } |
| 320 | |
| 321 | // sqlAlertSendTimeout bounds all keymail sends for one effect's alert, so a |
| 322 | // stalled mail service cannot hold the serial sweep for long. |
| 323 | const sqlAlertSendTimeout = 15 * time.Second |
| 324 | |
| 325 | // alertSQLDatabaseReplaced tells operators once per effect, through the same |
| 326 | // alert memory as box alerts: keymail where a grant exists, otherwise a log |
| 327 | // line. Sends run under the sweep's context. |
| 328 | func (s *Server) alertSQLDatabaseReplaced(ctx context.Context, account, app, effect string) { |
| 329 | rule := sqlReplacedRule(account, app, effect) |
| 330 | subject := fmt.Sprintf("[carlos] SQL effect %s on %s/%s needs the restore break-glass path", effect, account, app) |
| 331 | body := fmt.Sprintf("SQL effect %s on %s/%s is bound to a console database that has been replaced. It cannot be settled automatically.\nFollow the break-glass section of the console restore runbook.\n", effect, account, app) |
| 332 | var run *alertRun |
| 333 | if s.sendApp != nil { |
| 334 | r, err := s.alertRecipients() |
| 335 | if err != nil { |
| 336 | // Possibly transient: record nothing, so the next sweep tries |
| 337 | // again rather than filing the alert as undeliverable for good. |
| 338 | log.Printf("console: claim SQL alert: recipients: %v", err) |
| 339 | return |
| 340 | } |
| 341 | run = r |
| 342 | } |
| 343 | if run == nil || len(run.to) == 0 { |
| 344 | if s.alertDue(sqlReceiptAlertBox, rule, noMailboxAddress, true) { |
| 345 | log.Printf("console: ALERT %s", subject) |
| 346 | s.recordAlertSeen(sqlReceiptAlertBox, rule, noMailboxAddress, true) |
| 347 | } |
| 348 | return |
| 349 | } |
| 350 | ctx, cancel := context.WithTimeout(ctx, sqlAlertSendTimeout) |
| 351 | defer cancel() |
| 352 | for _, op := range run.to { |
| 353 | if !s.alertDue(sqlReceiptAlertBox, rule, op.address, true) { |
| 354 | continue |
| 355 | } |
| 356 | // Only when there is something to deliver, as in alertPass. |
| 357 | run.noteMissing() |
| 358 | _, err := s.sendApp(ctx, op.token, op.address, subject, body) |
| 359 | if err != nil { |
| 360 | log.Printf("console: claim SQL alert: keymail to %s: %v", op.address, err) |
| 361 | continue |
| 362 | } |
| 363 | s.recordAlertSeen(sqlReceiptAlertBox, rule, op.address, true) |
| 364 | } |
| 365 | } |
| 366 | |
amadan