| 1 | package hostbox |
| 2 | |
| 3 | import ( |
| 4 | "context" |
| 5 | "crypto/rand" |
| 6 | "crypto/sha256" |
| 7 | "database/sql" |
| 8 | "encoding/hex" |
| 9 | "encoding/json" |
| 10 | "errors" |
| 11 | "os" |
| 12 | "path/filepath" |
| 13 | "sync" |
| 14 | "time" |
| 15 | |
| 16 | "github.com/carlosframework/platform/internal/hostguard" |
| 17 | "github.com/carlosframework/platform/internal/rootstate" |
| 18 | ) |
| 19 | |
| 20 | var ErrLegacyJournalProof = errors.New("legacy journal requires root-signed stop evidence") |
| 21 | |
| 22 | // ErrJournalUncertain says nothing about whether a request committed. |
| 23 | var ErrJournalUncertain = errors.New("hostbox: journal request outcome uncertain") |
| 24 | |
| 25 | // A mutation executes only within JournalRequestWindow of the caller's issue |
| 26 | // time, so a receipt older than JournalReceiptRetention can never be needed: |
| 27 | // its request either replayed already or can only ever answer uncertain. |
| 28 | const ( |
| 29 | JournalRequestWindow = 15 * time.Minute |
| 30 | JournalReceiptRetention = 24 * time.Hour |
| 31 | journalPruneBatch = 256 |
| 32 | ) |
| 33 | |
| 34 | // journalPrepare fixes a request's identity once, before its first attempt. |
| 35 | // Reads get a fresh random ID. Every mutation is v2: a per-call nonce and issue time |
| 36 | // are hashed in, so two distinct calls with identical payloads never share a |
| 37 | // receipt. Retries resend the same prepared request. |
| 38 | func journalPrepare(req JournalRequest, now time.Time) (JournalRequest, error) { |
| 39 | req.Nonce, req.IssuedAt = "", 0 |
| 40 | switch { |
| 41 | case journalRead(req.Op): |
| 42 | var id [16]byte |
| 43 | if _, err := rand.Read(id[:]); err != nil { |
| 44 | return req, err |
| 45 | } |
| 46 | req.V = 1 |
| 47 | req.ID = hex.EncodeToString(id[:]) |
| 48 | default: |
| 49 | var nonce [16]byte |
| 50 | if _, err := rand.Read(nonce[:]); err != nil { |
| 51 | return req, err |
| 52 | } |
| 53 | req.V = 2 |
| 54 | if req.RecordKey != "" || req.RecordETag != "" { |
| 55 | req.V = 3 |
| 56 | } |
| 57 | if req.ServingPark != nil { |
| 58 | req.V = 4 |
| 59 | } |
| 60 | req.Nonce = hex.EncodeToString(nonce[:]) |
| 61 | req.IssuedAt = now.UnixNano() |
| 62 | req.ID = journalRequestID(req) |
| 63 | } |
| 64 | return req, nil |
| 65 | } |
| 66 | |
| 67 | // JournalRequest carries values, never SQL or caller-chosen filesystem paths. |
| 68 | // ID is retained by the transport across retries. Journal-only effects and |
| 69 | // their replies share a FULL-synchronous transaction. Parking records a pending |
| 70 | // receipt before filesystem work and a completed reply afterwards. |
| 71 | type JournalRequest struct { |
| 72 | Timer *TimerMutation `json:"timer,omitempty"` |
| 73 | ServingPark *ServingParkRequest `json:"serving_park,omitempty"` |
| 74 | RecordKey string `json:"record_key,omitempty"` |
| 75 | RecordETag string `json:"record_etag,omitempty"` |
| 76 | Revision uint64 `json:"inventory_revision"` |
| 77 | V int `json:"v"` |
| 78 | ID string `json:"request_id"` |
| 79 | Op string `json:"operation"` |
| 80 | Target hostguard.FenceTarget `json:"target"` |
| 81 | Launch LaunchInventory `json:"launch"` |
| 82 | Reservation LaunchReservation `json:"reservation"` |
| 83 | File LaunchFile `json:"file"` |
| 84 | Producer string `json:"producer"` |
| 85 | Previous ProducerCompletion `json:"previous"` |
| 86 | PID int `json:"pid"` |
| 87 | ProcessStart string `json:"process_start"` |
| 88 | Inventory StopInventory `json:"inventory"` |
| 89 | Nonce string `json:"nonce,omitempty"` |
| 90 | IssuedAt int64 `json:"issued_at,omitempty"` |
| 91 | } |
| 92 | type JournalResponse struct { |
| 93 | ServingPark *ServingParkResult `json:"serving_park,omitempty"` |
| 94 | FenceView *hostguard.FenceView `json:"fence_view,omitempty"` |
| 95 | Revision uint64 `json:"inventory_revision"` |
| 96 | LastRequest string `json:"last_request"` |
| 97 | V int `json:"v"` |
| 98 | ID string `json:"request_id"` |
| 99 | Launch LaunchInventory `json:"launch"` |
| 100 | Launches []LaunchInventory `json:"launches"` |
| 101 | Inventory StopInventory `json:"inventory"` |
| 102 | Completion ProducerCompletion `json:"completion"` |
| 103 | Found bool `json:"found"` |
| 104 | Error string `json:"error"` |
| 105 | } |
| 106 | type JournalCall func(context.Context, JournalRequest) (JournalResponse, error) |
| 107 | |
| 108 | // JournalUncertainError retains the exact request for explicit same-ID replay. |
| 109 | // Read-back can inform recovery; it never proves that a pending write failed. |
| 110 | type JournalUncertainError struct { |
| 111 | Request JournalRequest |
| 112 | Cause error |
| 113 | } |
| 114 | |
| 115 | func (e *JournalUncertainError) Error() string { return "journal request outcome uncertain" } |
| 116 | func (e *JournalUncertainError) Unwrap() error { return e.Cause } |
| 117 | |
| 118 | func RemoteJournals(call JournalCall) (*LaunchJournal, *StopInventoryDB) { |
| 119 | return &LaunchJournal{remote: call}, &StopInventoryDB{remote: call} |
| 120 | } |
| 121 | func journalCall(ctx context.Context, call JournalCall, req JournalRequest) (JournalResponse, error) { |
| 122 | req, err := journalPrepare(req, time.Now()) |
| 123 | if err != nil { |
| 124 | return JournalResponse{}, err |
| 125 | } |
| 126 | return journalSend(ctx, call, req, false) |
| 127 | } |
| 128 | |
| 129 | // ReplayUncertain resends the exact request an uncertain call retained. The |
| 130 | // earlier attempt may have committed, so the replay stays uncertain on every |
| 131 | // refusal, including after the receipt was pruned. Only a reply that names |
| 132 | // this exact request with no error confirms it. |
| 133 | func (j *LaunchJournal) ReplayUncertain(ctx context.Context, uncertain *JournalUncertainError) (JournalResponse, error) { |
| 134 | if j == nil || j.remote == nil || uncertain == nil || uncertain.Request.ID == "" { |
| 135 | return JournalResponse{}, ErrUnsupported |
| 136 | } |
| 137 | return journalSend(ctx, j.remote, uncertain.Request, true) |
| 138 | } |
| 139 | |
| 140 | func journalSend(ctx context.Context, call JournalCall, req JournalRequest, replay bool) (JournalResponse, error) { |
| 141 | out, err := call(ctx, req) |
| 142 | if err != nil { |
| 143 | return out, &JournalUncertainError{Request: req, Cause: err} |
| 144 | } |
| 145 | if out.V != 1 || out.ID != req.ID { |
| 146 | return out, &JournalUncertainError{Request: req, Cause: ErrJournalUncertain} |
| 147 | } |
| 148 | switch out.Error { |
| 149 | case "": |
| 150 | return out, nil |
| 151 | } |
| 152 | if replay { |
| 153 | return out, &JournalUncertainError{Request: req, Cause: ErrJournalUncertain} |
| 154 | } |
| 155 | switch out.Error { |
| 156 | case "missing": |
| 157 | return out, sql.ErrNoRows |
| 158 | case "changed": |
| 159 | return out, ErrChanged |
| 160 | case "unsupported": |
| 161 | return out, ErrUnsupported |
| 162 | case "uncertain": |
| 163 | return out, &JournalUncertainError{Request: req, Cause: ErrJournalUncertain} |
| 164 | default: |
| 165 | return out, ErrTypedInventory |
| 166 | } |
| 167 | } |
| 168 | |
| 169 | // JournalReadOrSettle identifies read-only operations and positive completion |
| 170 | // of already issued work. It never includes creating a new producer or role. |
| 171 | func JournalReadOrSettle(op string) bool { |
| 172 | return journalRead(op) || op == "close-writers" || op == "started" || op == "settled" || op == "complete" |
| 173 | } |
| 174 | |
| 175 | func journalRead(op string) bool { |
| 176 | return op == "fence-view" || op == "read" || op == "session" || op == "has-host" || op == "has-effects" || op == "inventory-read" || op == "inventory-exact" || op == "inventory-version" |
| 177 | } |
| 178 | |
| 179 | // journalRequestV1 freezes every field, including nested inventories, in an |
| 180 | // ordered v1 tuple. Do not extend these tuples when the public structs grow: |
| 181 | // a new hashed field requires a new protocol version and a separate encoder. |
| 182 | func journalRequestV1(q JournalRequest) []byte { |
| 183 | target := func(t hostguard.FenceTarget) any { |
| 184 | return []any{t.Region, t.ClaimGen, t.ObligationID, t.FenceID, t.Host, t.Incarnation, t.Stamp, t.Session, t.Op, t.Phase, t.Box, t.Enrollment, t.BootID, t.ControllerID, t.ClaimETag, t.DirectoryRevision} |
| 185 | } |
| 186 | file := func(f LaunchFile) any { return []any{f.Device, f.Inode} } |
| 187 | group := func(g LaunchGroup) any { return []any{g.RelativePath, g.Device, g.Inode, g.InvocationID} } |
| 188 | reservation := func(r LaunchReservation) any { |
| 189 | return []any{r.ID, r.Role, r.InvocationID, r.Inputs, r.State, r.ProcessStart, target(r.Target), group(r.Group), group(r.Parent), file(r.Executable), r.PID} |
| 190 | } |
| 191 | groups := make([]any, len(q.Launch.Groups)) |
| 192 | for i, g := range q.Launch.Groups { |
| 193 | groups[i] = group(g) |
| 194 | } |
| 195 | files := make([]any, len(q.Launch.Files)) |
| 196 | for i, f := range q.Launch.Files { |
| 197 | files[i] = file(f) |
| 198 | } |
| 199 | reservations := make([]any, len(q.Launch.Reservations)) |
| 200 | for i, r := range q.Launch.Reservations { |
| 201 | reservations[i] = reservation(r) |
| 202 | } |
| 203 | launch := []any{q.Launch.Completed.Count, q.Launch.Completed.Digest, q.Launch.ControllerPID, q.Launch.ProcessStart, target(q.Launch.Target), groups, files, append([]string{}, q.Launch.ProducerIDs...), q.Launch.NoState, reservations} |
| 204 | entries := make([]any, len(q.Inventory.Entries)) |
| 205 | for i, e := range q.Inventory.Entries { |
| 206 | entries[i] = []any{e.Device, e.Inode, e.Type, e.Kind, e.ArtifactID, e.ManagedRoot, e.RelativePath, e.RowIdentity, e.Ownership, e.Materialization} |
| 207 | } |
| 208 | inventory := []any{q.Inventory.Complete, q.Inventory.Host, q.Inventory.Incarnation, q.Inventory.InstallationStamp, q.Inventory.Box, q.Inventory.Enrollment, q.Inventory.DirectoryRevision, entries} |
| 209 | raw, _ := json.Marshal([]any{"carlos-journal-request-v1", q.Revision, q.Op, target(q.Target), launch, reservation(q.Reservation), file(q.File), q.Producer, q.Previous.Count, q.Previous.Digest, q.PID, q.ProcessStart, inventory}) |
| 210 | return raw |
| 211 | } |
| 212 | |
| 213 | // journalRequestV2 is the frozen v1 tuple plus the caller's per-call nonce and |
| 214 | // issue time. It is a separate encoder; v1 identities are unchanged. |
| 215 | func journalRequestV2(q JournalRequest) []byte { |
| 216 | raw, _ := json.Marshal([]any{"carlos-journal-request-v2", string(journalRequestV1(q)), q.Nonce, q.IssuedAt}) |
| 217 | return raw |
| 218 | } |
| 219 | func journalRequestHash(req JournalRequest) [sha256.Size]byte { |
| 220 | identities := make([]rootstate.FileIdentity, len(req.Inventory.Entries)) |
| 221 | present := false |
| 222 | for i, e := range req.Inventory.Entries { |
| 223 | identities[i] = e.Identity |
| 224 | present = present || e.Identity != (rootstate.FileIdentity{}) |
| 225 | } |
| 226 | if present { |
| 227 | base := req |
| 228 | base.Inventory.Entries = append([]StopInventoryEntry(nil), req.Inventory.Entries...) |
| 229 | for i := range base.Inventory.Entries { |
| 230 | base.Inventory.Entries[i].Identity = rootstate.FileIdentity{} |
| 231 | } |
| 232 | raw, _ := json.Marshal([]any{"carlos-inventory-identity-v1", journalRequestHash(base), identities}) |
| 233 | return sha256.Sum256(raw) |
| 234 | } |
| 235 | |
| 236 | if req.V == 4 { |
| 237 | raw, _ := json.Marshal([]any{"carlos-journal-request-v4", string(journalRequestV2(req)), journalServingParkV4(req.ServingPark)}) |
| 238 | return sha256.Sum256(raw) |
| 239 | } |
| 240 | if req.V == 3 { |
| 241 | raw, _ := json.Marshal([]any{"carlos-journal-request-v3", string(journalRequestV2(req)), req.RecordKey, req.RecordETag}) |
| 242 | return sha256.Sum256(raw) |
| 243 | } |
| 244 | if req.V == 2 { |
| 245 | return sha256.Sum256(journalRequestV2(req)) |
| 246 | } |
| 247 | return sha256.Sum256(journalRequestV1(req)) |
| 248 | } |
| 249 | func journalRequestID(req JournalRequest) string { |
| 250 | sum := journalRequestHash(req) |
| 251 | return hex.EncodeToString(sum[:16]) |
| 252 | } |
| 253 | func journalRequestDigest(req JournalRequest) string { |
| 254 | sum := journalRequestHash(req) |
| 255 | return hex.EncodeToString(sum[:]) |
| 256 | } |
| 257 | func validateJournalRequest(q JournalRequest) error { |
| 258 | if q.V == 4 { |
| 259 | return validateServingParkRequest(q) |
| 260 | } |
| 261 | if q.ServingPark != nil || ServingParkOperation(q.Op) { |
| 262 | return ErrTypedJournal |
| 263 | } |
| 264 | // The frozen request hash does not carry this root-only response field. |
| 265 | if q.Launch.WritersClosed || (q.V != 3 && (q.RecordKey != "" || q.RecordETag != "")) || (q.V == 3 && (q.Op != "inventory-write" || q.RecordKey == "" || q.RecordETag == "")) { |
| 266 | return ErrTypedJournal |
| 267 | } |
| 268 | id, err := hex.DecodeString(q.ID) |
| 269 | if err != nil || len(id) != 16 { |
| 270 | return ErrTypedJournal |
| 271 | } |
| 272 | switch { |
| 273 | case journalRead(q.Op): |
| 274 | if q.V != 1 || q.Nonce != "" || q.IssuedAt != 0 { |
| 275 | return ErrTypedJournal |
| 276 | } |
| 277 | return nil |
| 278 | default: |
| 279 | nonce, err := hex.DecodeString(q.Nonce) |
| 280 | if (q.V != 2 && q.V != 3) || err != nil || len(nonce) != 16 || q.IssuedAt <= 0 { |
| 281 | return ErrTypedJournal |
| 282 | } |
| 283 | } |
| 284 | if q.ID != journalRequestID(q) { |
| 285 | return ErrChanged |
| 286 | } |
| 287 | return nil |
| 288 | } |
| 289 | |
| 290 | type journalRequestKey struct{} |
| 291 | |
| 292 | const inventoryVersionsSQL = "CREATE TABLE IF NOT EXISTS journal_inventory_versions(host TEXT NOT NULL, incarnation TEXT NOT NULL, stamp INTEGER NOT NULL, revision INTEGER NOT NULL CHECK(revision>0 AND typeof(revision)='integer'), request TEXT NOT NULL, PRIMARY KEY(host,incarnation,stamp))" |
| 293 | |
| 294 | type journalExecutor interface { |
| 295 | ExecContext(context.Context, string, ...any) (sql.Result, error) |
| 296 | QueryContext(context.Context, string, ...any) (*sql.Rows, error) |
| 297 | QueryRowContext(context.Context, string, ...any) *sql.Row |
| 298 | } |
| 299 | type journalTransactionKey struct{} |
| 300 | |
| 301 | func journalExecutorFor(ctx context.Context, db *sql.DB) journalExecutor { |
| 302 | if tx, ok := ctx.Value(journalTransactionKey{}).(*sql.Tx); ok { |
| 303 | return tx |
| 304 | } |
| 305 | return db |
| 306 | } |
| 307 | func journalTransaction(ctx context.Context, db *sql.DB) (*sql.Tx, func() error, func() error, error) { |
| 308 | if tx, ok := ctx.Value(journalTransactionKey{}).(*sql.Tx); ok { |
| 309 | return tx, func() error { return nil }, func() error { return nil }, nil |
| 310 | } |
| 311 | tx, err := db.BeginTx(ctx, nil) |
| 312 | if err != nil { |
| 313 | return nil, nil, nil, err |
| 314 | } |
| 315 | return tx, tx.Commit, tx.Rollback, nil |
| 316 | } |
| 317 | func journalIdentity(path string, original os.FileInfo) error { |
| 318 | if original == nil { |
| 319 | return nil |
| 320 | } |
| 321 | now, err := os.Lstat(path) |
| 322 | if err != nil { |
| 323 | return err |
| 324 | } |
| 325 | if !now.Mode().IsRegular() || !os.SameFile(original, now) { |
| 326 | return ErrChanged |
| 327 | } |
| 328 | return nil |
| 329 | } |
| 330 | |
| 331 | // JournalService owns one protected database and serializes a request's replay |
| 332 | // check, typed mutation and receipt. Local root recovery uses the same DB. A |
| 333 | // replay never repeats an effect, even when the reply or root process was lost. |
| 334 | // A v2 mutation executes only within JournalRequestWindow of its issue time, |
| 335 | // so once a receipt is older than JournalReceiptRetention no request can |
| 336 | // still need it: a replay inside the window found it, and anything later can |
| 337 | // only answer uncertain. Mutations prune such settled receipts in bounded |
| 338 | // batches. inventory-write needs no receipt to replay: its version row names |
| 339 | // the last committed request. |
| 340 | type JournalService struct { |
| 341 | parkGate chan struct{} |
| 342 | Launches *LaunchJournal |
| 343 | Inventory *StopInventoryDB |
| 344 | Now func() time.Time |
| 345 | mu sync.Mutex |
| 346 | } |
| 347 | |
| 348 | func NewJournalService(j *LaunchJournal) (*JournalService, error) { |
| 349 | if j == nil || j.db == nil || j.remote != nil { |
| 350 | return nil, ErrUnsupported |
| 351 | } |
| 352 | for _, q := range []string{inventoryVersionsSQL, |
| 353 | "CREATE TABLE IF NOT EXISTS stop_inventory(host TEXT NOT NULL,incarnation TEXT NOT NULL,stamp INTEGER NOT NULL,body BLOB NOT NULL,PRIMARY KEY(host,incarnation,stamp))", |
| 354 | "CREATE TABLE IF NOT EXISTS journal_requests(id TEXT PRIMARY KEY, peer TEXT NOT NULL, digest TEXT NOT NULL, response BLOB NOT NULL, created INTEGER NOT NULL DEFAULT 0, issued INTEGER NOT NULL DEFAULT 0)", |
| 355 | "CREATE TABLE IF NOT EXISTS journal_receipt_retention(id INTEGER PRIMARY KEY CHECK(id=1), reject_before INTEGER NOT NULL)", |
| 356 | "INSERT INTO journal_receipt_retention(id,reject_before) VALUES(1,0) ON CONFLICT(id) DO NOTHING", |
| 357 | } { |
| 358 | if _, err := j.db.Exec(q); err != nil { |
| 359 | return nil, err |
| 360 | } |
| 361 | } |
| 362 | // Receipts written before retention existed carry created=0 and are pruned |
| 363 | // first; their v1 mutation IDs can no longer execute. |
| 364 | var created int |
| 365 | if err := j.db.QueryRow("SELECT count(*) FROM pragma_table_info('journal_requests') WHERE name='created'").Scan(&created); err != nil { |
| 366 | return nil, err |
| 367 | } |
| 368 | if created == 0 { |
| 369 | if _, err := j.db.Exec("ALTER TABLE journal_requests ADD COLUMN created INTEGER NOT NULL DEFAULT 0"); err != nil { |
| 370 | return nil, err |
| 371 | } |
| 372 | } |
| 373 | var issued int |
| 374 | if err := j.db.QueryRow("SELECT count(*) FROM pragma_table_info('journal_requests') WHERE name='issued'").Scan(&issued); err != nil { |
| 375 | return nil, err |
| 376 | } |
| 377 | if issued == 0 { |
| 378 | if _, err := j.db.Exec("ALTER TABLE journal_requests ADD COLUMN issued INTEGER NOT NULL DEFAULT 0"); err != nil { |
| 379 | return nil, err |
| 380 | } |
| 381 | } |
| 382 | if _, err := j.db.Exec("CREATE INDEX IF NOT EXISTS journal_requests_created ON journal_requests(created)"); err != nil { |
| 383 | return nil, err |
| 384 | } |
| 385 | dir, err := os.Open(filepath.Dir(j.path)) |
| 386 | if err != nil { |
| 387 | return nil, err |
| 388 | } |
| 389 | err = dir.Sync() |
| 390 | closeErr := dir.Close() |
| 391 | if err != nil || closeErr != nil { |
| 392 | return nil, errors.Join(err, closeErr) |
| 393 | } |
| 394 | return &JournalService{parkGate: make(chan struct{}, 1), Launches: j, Inventory: &StopInventoryDB{db: j.db, shared: true}}, nil |
| 395 | } |
| 396 | |
| 397 | // Replay returns only a byte-identical request by the same authenticated owner. |
| 398 | func (s *JournalService) Replay(ctx context.Context, peer string, q JournalRequest) (JournalResponse, bool, error) { |
| 399 | out := JournalResponse{V: 1, ID: q.ID} |
| 400 | if err := validateJournalRequest(q); err != nil { |
| 401 | return out, false, err |
| 402 | } |
| 403 | if journalRead(q.Op) { |
| 404 | return out, false, nil |
| 405 | } |
| 406 | digestExpected := journalRequestDigest(q) |
| 407 | var priorPeer, digest string |
| 408 | var body []byte |
| 409 | err := s.Launches.db.QueryRowContext(ctx, "SELECT peer,digest,response FROM journal_requests WHERE id=?", q.ID).Scan(&priorPeer, &digest, &body) |
| 410 | if errors.Is(err, sql.ErrNoRows) { |
| 411 | return out, false, nil |
| 412 | } |
| 413 | if err != nil { |
| 414 | return out, false, err |
| 415 | } |
| 416 | if priorPeer != peer || digest != digestExpected { |
| 417 | return out, false, ErrChanged |
| 418 | } |
| 419 | err = json.Unmarshal(body, &out) |
| 420 | return out, true, err |
| 421 | } |
| 422 | |
| 423 | func (s *JournalService) Execute(ctx context.Context, peer string, q JournalRequest) (JournalResponse, error) { |
| 424 | return s.execute(ctx, peer, q, false) |
| 425 | } |
| 426 | func (s *JournalService) CompletePendingInventory(ctx context.Context, peer string, q JournalRequest) (JournalResponse, error) { |
| 427 | if q.V != 3 || q.Op != "inventory-write" { |
| 428 | return JournalResponse{}, ErrTypedJournal |
| 429 | } |
| 430 | return s.execute(ctx, peer, q, true) |
| 431 | } |
| 432 | func (s *JournalService) ExecuteRootInventory(ctx context.Context, peer string, q JournalRequest, in StopInventory, pending bool) (JournalResponse, error) { |
| 433 | if q.V != 3 || q.Op != "inventory-write" || q.Inventory.Host != in.Host || q.Inventory.Incarnation != in.Incarnation || q.Inventory.InstallationStamp != in.InstallationStamp { |
| 434 | return JournalResponse{}, ErrTypedJournal |
| 435 | } |
| 436 | return s.executeWithInventory(ctx, peer, q, pending, &in) |
| 437 | } |
| 438 | |
| 439 | func (s *JournalService) execute(ctx context.Context, peer string, q JournalRequest, pendingInventory bool) (JournalResponse, error) { |
| 440 | return s.executeWithInventory(ctx, peer, q, pendingInventory, nil) |
| 441 | } |
| 442 | func (s *JournalService) executeWithInventory(ctx context.Context, peer string, q JournalRequest, pendingInventory bool, inventory *StopInventory) (JournalResponse, error) { |
| 443 | out := JournalResponse{V: 1, ID: q.ID} |
| 444 | if err := validateJournalRequest(q); err != nil { |
| 445 | return out, err |
| 446 | } |
| 447 | if peer == "" { |
| 448 | return out, ErrTypedJournal |
| 449 | } |
| 450 | if journalRead(q.Op) { |
| 451 | return s.read(ctx, q) |
| 452 | } |
| 453 | digest := journalRequestDigest(q) |
| 454 | if inventory != nil { |
| 455 | q.Inventory = *inventory |
| 456 | } |
| 457 | s.mu.Lock() |
| 458 | defer s.mu.Unlock() |
| 459 | tx, err := s.Launches.db.BeginTx(ctx, nil) |
| 460 | if err != nil { |
| 461 | return out, err |
| 462 | } |
| 463 | defer tx.Rollback() |
| 464 | var priorPeer, priorDigest string |
| 465 | var prior []byte |
| 466 | err = tx.QueryRowContext(ctx, "SELECT peer,digest,response FROM journal_requests WHERE id=?", q.ID).Scan(&priorPeer, &priorDigest, &prior) |
| 467 | if err == nil { |
| 468 | if priorPeer != peer || priorDigest != digest { |
| 469 | return out, ErrChanged |
| 470 | } |
| 471 | err = json.Unmarshal(prior, &out) |
| 472 | return out, err |
| 473 | } |
| 474 | if !errors.Is(err, sql.ErrNoRows) { |
| 475 | return out, err |
| 476 | } |
| 477 | // Inventory's permanent revision row can prove this exact nonce committed |
| 478 | // even after the receipt expired. It never authorizes another execution. |
| 479 | ctx = context.WithValue(ctx, journalTransactionKey{}, tx) |
| 480 | if q.Op == "inventory-write" { |
| 481 | prior, e := s.Inventory.inventoryVersion(ctx, hostguard.FenceTarget{Host: q.Inventory.Host, Incarnation: q.Inventory.Incarnation, Stamp: q.Inventory.InstallationStamp}) |
| 482 | if e != nil { |
| 483 | return out, e |
| 484 | } |
| 485 | if prior.Revision == q.Revision+1 && prior.LastRequest == q.ID { |
| 486 | out.Revision = prior.Revision |
| 487 | return out, nil |
| 488 | } |
| 489 | } |
| 490 | now := time.Now() |
| 491 | if s.Now != nil { |
| 492 | now = s.Now() |
| 493 | } |
| 494 | var rejectBefore int64 |
| 495 | if err = tx.QueryRowContext(ctx, "SELECT reject_before FROM journal_receipt_retention WHERE id=1").Scan(&rejectBefore); err != nil { |
| 496 | return out, err |
| 497 | } |
| 498 | // The rejection boundary only ever rises to the issue time of a receipt |
| 499 | // this transaction deletes, so a pruned request can never execute again |
| 500 | // whatever the clock does, and a forward clock jump cannot raise it past |
| 501 | // real time: it covers only requests that actually committed. |
| 502 | if (q.V == 2 || q.V == 3) && !pendingInventory { |
| 503 | age := now.Sub(time.Unix(0, q.IssuedAt)) |
| 504 | if q.IssuedAt <= rejectBefore || age > JournalRequestWindow || age < -JournalRequestWindow { |
| 505 | return out, ErrJournalUncertain |
| 506 | } |
| 507 | } |
| 508 | rejectBefore, err = journalPrune(ctx, tx, rejectBefore, now) |
| 509 | if err != nil { |
| 510 | return out, err |
| 511 | } |
| 512 | ctx = context.WithValue(ctx, journalTransactionKey{}, tx) |
| 513 | ctx = context.WithValue(ctx, journalRequestKey{}, q.ID) |
| 514 | switch q.Op { |
| 515 | case "close-writers": |
| 516 | err = s.Launches.CloseWriters(ctx, q.Target) |
| 517 | case "create": |
| 518 | err = s.Launches.Create(ctx, q.Launch) |
| 519 | case "prelaunch": |
| 520 | err = s.Launches.Prelaunch(ctx, q.Launch) |
| 521 | case "intend": |
| 522 | err = s.Launches.IntendRole(ctx, q.Reservation) |
| 523 | case "reserve": |
| 524 | err = s.Launches.Reserve(ctx, q.Reservation) |
| 525 | case "started": |
| 526 | err = s.Launches.Started(ctx, q.Reservation, q.PID, q.ProcessStart) |
| 527 | case "settled": |
| 528 | err = s.Launches.Settled(ctx, q.Reservation) |
| 529 | case "file": |
| 530 | err = s.Launches.RecordFile(ctx, q.Target, q.File) |
| 531 | case "producer": |
| 532 | err = s.Launches.RecordProducer(ctx, q.Target, q.Producer) |
| 533 | case "complete": |
| 534 | out.Completion, err = s.Launches.CompleteProducer(ctx, q.Target, q.Producer, q.Previous) |
| 535 | case "inventory-write": |
| 536 | prior, e := s.Inventory.inventoryVersion(ctx, hostguard.FenceTarget{Host: q.Inventory.Host, Incarnation: q.Inventory.Incarnation, Stamp: q.Inventory.InstallationStamp}) |
| 537 | if e != nil { |
| 538 | return out, e |
| 539 | } |
| 540 | if prior.Revision != q.Revision { |
| 541 | return out, ErrChanged |
| 542 | } |
| 543 | err = s.Inventory.WriteStopInventory(ctx, q.Inventory) |
| 544 | out.Revision = prior.Revision + 1 |
| 545 | default: |
| 546 | return out, ErrUnsupported |
| 547 | } |
| 548 | // A failed operation is rolled back, so no partial mutation acquires a receipt. |
| 549 | if err != nil { |
| 550 | return out, err |
| 551 | } |
| 552 | body, err := json.Marshal(out) |
| 553 | if err != nil { |
| 554 | return out, err |
| 555 | } |
| 556 | if _, err = tx.ExecContext(ctx, "INSERT INTO journal_requests(id,peer,digest,response,created,issued) VALUES(?,?,?,?,?,?)", q.ID, peer, digest, body, now.UnixNano(), q.IssuedAt); err != nil { |
| 557 | return out, err |
| 558 | } |
| 559 | if err = tx.Commit(); err != nil { |
| 560 | return out, err |
| 561 | } |
| 562 | return out, nil |
| 563 | } |
| 564 | |
| 565 | // Reads neither take a write transaction nor allocate durable request receipts. |
| 566 | // inventory-version is a single joined SELECT, so revision/body/last ID form |
| 567 | // one SQLite snapshot even when a root writer commits concurrently. |
| 568 | func (s *JournalService) read(ctx context.Context, q JournalRequest) (JournalResponse, error) { |
| 569 | out := JournalResponse{V: 1, ID: q.ID} |
| 570 | var err error |
| 571 | switch q.Op { |
| 572 | case "read": |
| 573 | out.Launch, err = s.Launches.Read(ctx, q.Target) |
| 574 | case "session": |
| 575 | out.Launches, err = s.Launches.SessionLaunches(ctx, q.Target.Host, q.Target.Incarnation, q.Target.Stamp, q.Target.Session, q.Target.Box, q.Target.Enrollment) |
| 576 | case "has-host": |
| 577 | out.Found, err = s.Launches.HasHost(ctx, q.Target.Host) |
| 578 | case "has-effects": |
| 579 | out.Found, err = s.Launches.HasInstallationEffects(ctx, q.Target.Host, q.Target.Incarnation, q.Target.Stamp) |
| 580 | case "inventory-version": |
| 581 | out, err = s.Inventory.inventoryVersion(ctx, q.Target) |
| 582 | out.V = 1 |
| 583 | out.ID = q.ID |
| 584 | case "inventory-read": |
| 585 | out.Inventory, err = s.Inventory.ReadStopInventory(ctx, q.Target.Host) |
| 586 | case "inventory-exact": |
| 587 | out.Inventory, err = s.Inventory.ReadExactStopInventory(ctx, q.Target) |
| 588 | default: |
| 589 | return out, ErrUnsupported |
| 590 | } |
| 591 | return out, err |
| 592 | } |
| 593 | |
| 594 | // journalPrune deletes one bounded batch of receipts past retention and raises |
| 595 | // the persisted boundary to cover their requests. A request executed within |
| 596 | // JournalRequestWindow of its issue time, so an unrecorded issue time is at |
| 597 | // most that far past the receipt's creation. |
| 598 | func journalPrune(ctx context.Context, tx *sql.Tx, rejectBefore int64, now time.Time) (int64, error) { |
| 599 | cutoff := max(rejectBefore-int64(JournalRequestWindow), now.Add(-JournalReceiptRetention).UnixNano()) |
| 600 | rows, err := tx.QueryContext(ctx, "SELECT rowid,created,issued FROM journal_requests WHERE created < ? ORDER BY created LIMIT ?", cutoff, journalPruneBatch) |
| 601 | if err != nil { |
| 602 | return rejectBefore, err |
| 603 | } |
| 604 | var ids []int64 |
| 605 | covered := rejectBefore |
| 606 | for rows.Next() { |
| 607 | var id, created, issued int64 |
| 608 | if err = rows.Scan(&id, &created, &issued); err != nil { |
| 609 | rows.Close() |
| 610 | return rejectBefore, err |
| 611 | } |
| 612 | if issued <= 0 { |
| 613 | issued = created + int64(JournalRequestWindow) |
| 614 | } |
| 615 | ids = append(ids, id) |
| 616 | covered = max(covered, issued) |
| 617 | } |
| 618 | if err = errors.Join(rows.Err(), rows.Close()); err != nil { |
| 619 | return rejectBefore, err |
| 620 | } |
| 621 | if len(ids) == 0 { |
| 622 | return rejectBefore, nil |
| 623 | } |
| 624 | if _, err = tx.ExecContext(ctx, "UPDATE journal_receipt_retention SET reject_before=? WHERE id=1", covered); err != nil { |
| 625 | return rejectBefore, err |
| 626 | } |
| 627 | for _, id := range ids { |
| 628 | if _, err = tx.ExecContext(ctx, "DELETE FROM journal_requests WHERE rowid=?", id); err != nil { |
| 629 | return rejectBefore, err |
| 630 | } |
| 631 | } |
| 632 | return covered, nil |
| 633 | } |
| 634 | |
amadan