1// Package appstate implements the out.AppState boundary on bbolt:
2// one state.db file with per-app buckets, single-writer transactions,
3// and crash-safe commit semantics from bbolt itself. It replaces the
4// prepared multi-file JSON adapter (no compatibility layer).
5// It never reads secret values and never touches OCI manifest storage.
6package appstate
7
8import (
9 "context"
10 "encoding/json"
11 "errors"
12 "fmt"
13 "os"
14 "path/filepath"
15 "sort"
16 "time"
17
18 "github.com/bnema/zerowrap"
19 "github.com/google/uuid"
20 bolt "go.etcd.io/bbolt"
21 bolterrors "go.etcd.io/bbolt/errors"
22
23 "github.com/bnema/gordon/internal/domain"
24)
25
26// Bucket layout inside state.db.
27var (
28 bucketMeta = []byte("meta")
29 bucketCheckpoint = []byte("checkpoint")
30 bucketApps = []byte("apps")
31 bucketOwnershipHistory = []byte("ownership_history")
32 keyStoreRecord = []byte("store")
33 keyAppRecord = []byte("app")
34 keyDesired = []byte("desired")
35 keyActive = []byte("active")
36 keyIntent = []byte("intent")
37 keyOwnership = []byte("ownership")
38 keyInhibitions = []byte("recovery_inhibitions")
39 keyOwnershipMigration = []byte("ownership_history_v1")
40 subRevisions = []byte("revisions")
41 subIntents = []byte("intents")
42 subOps = []byte("ops")
43)
44
45// Store layout constants.
46const (
47 appsDir = "apps"
48 stateFile = "state.db"
49 dirPerm = 0o750
50)
51
52// appRecord tracks the stable internal UUID for one public app name.
53type appRecord struct {
54 ID string `json:"id"`
55 Name string `json:"name"`
56}
57
58// Store is a bbolt-backed out.AppState implementation.
59type Store struct {
60 db *bolt.DB
61 log zerowrap.Logger
62 // retention bounds unreferenced-revision GC per app. Zero means
63 // the compiled default (AppDefaultRevisionRetention).
64 retention int
65}
66
67// WithRevisionRetention overrides the unreferenced-revision GC bound.
68// Values <= 0 restore the compiled default.
69func (s *Store) WithRevisionRetention(n int) *Store {
70 if n > 0 {
71 s.retention = n
72 } else {
73 s.retention = 0
74 }
75 return s
76}
77
78// Retention returns the effective unreferenced-revision GC bound.
79func (s *Store) Retention() int {
80 if s.retention > 0 {
81 return s.retention
82 }
83 return domain.AppDefaultRevisionRetention
84}
85
86// NewStore creates the store rooted at dataDir (…/apps/state.db).
87func NewStore(dataDir string, log zerowrap.Logger) (*Store, error) {
88 root := filepath.Join(dataDir, appsDir)
89 if err := os.MkdirAll(root, dirPerm); err != nil {
90 return nil, fmt.Errorf("appstate: create store root: %w", domain.ErrAppStateIO)
91 }
92 db, err := bolt.Open(filepath.Join(root, stateFile), 0o600, &bolt.Options{Timeout: 5 * time.Second})
93 if err != nil {
94 return nil, fmt.Errorf("appstate: open state db: %w", domain.ErrAppStateIO)
95 }
96 store := &Store{db: db, log: log}
97 if err := store.init(); err != nil {
98 _ = db.Close()
99 return nil, err
100 }
101 return store, nil
102}
103
104// Close releases the underlying database handle.
105func (s *Store) Close() error {
106 if s == nil || s.db == nil {
107 return nil
108 }
109 if err := s.db.Close(); err != nil {
110 return fmt.Errorf("appstate: close state db: %w", domain.ErrAppStateIO)
111 }
112 return nil
113}
114
115// init ensures top-level buckets and the default checkpoint exist.
116func (s *Store) init() error {
117 return s.db.Update(func(tx *bolt.Tx) error {
118 if _, err := tx.CreateBucketIfNotExists(bucketMeta); err != nil {
119 return fmt.Errorf("appstate: create meta bucket: %w", domain.ErrAppStateIO)
120 }
121 checkpointBucket, err := tx.CreateBucketIfNotExists(bucketCheckpoint)
122 if err != nil {
123 return fmt.Errorf("appstate: create checkpoint bucket: %w", domain.ErrAppStateIO)
124 }
125 if checkpointBucket.Get(keyStoreRecord) == nil {
126 raw, err := marshal(domain.AppStoreCheckpoint{Version: domain.AppStoreVersion})
127 if err != nil {
128 return err
129 }
130 if err := checkpointBucket.Put(keyStoreRecord, raw); err != nil {
131 return fmt.Errorf("appstate: init checkpoint: %w", domain.ErrAppStateIO)
132 }
133 }
134 if _, err := tx.CreateBucketIfNotExists(bucketApps); err != nil {
135 return fmt.Errorf("appstate: create apps bucket: %w", domain.ErrAppStateIO)
136 }
137 return migrateOwnershipHistoryLocked(tx)
138 })
139}
140
141// migrateOwnershipHistoryLocked populates the incarnation-indexed
142// ownership history once. A reused app name never overwrites the
143// history of an earlier incarnation, so retained resources stay
144// protected after a remove/recreate. Existing records are copied
145// verbatim: claimsForOwnership maps every unrecognized lifecycle state
146// conservatively to retained.
147func migrateOwnershipHistoryLocked(tx *bolt.Tx) error {
148 meta := tx.Bucket(bucketMeta)
149 if meta == nil {
150 return fmt.Errorf("appstate: missing meta bucket: %w", domain.ErrAppStateIO)
151 }
152 history, err := tx.CreateBucketIfNotExists(bucketOwnershipHistory)
153 if err != nil {
154 return fmt.Errorf("appstate: create ownership history bucket: %w", domain.ErrAppStateIO)
155 }
156 if meta.Get(keyOwnershipMigration) != nil {
157 return nil
158 }
159
160 apps := tx.Bucket(bucketApps)
161 if apps != nil {
162 if err := apps.ForEach(func(name, _ []byte) error {
163 bucket := apps.Bucket(name)
164 if bucket == nil {
165 return nil
166 }
167 var ownership domain.AppOwnership
168 found, err := getRecord(bucket, keyOwnership, &ownership, "ownership")
169 if err != nil {
170 // A corrupt record cannot be migrated. It stays unwritten
171 // and surfaces as an inventory gap on every snapshot.
172 return nil
173 }
174 if !found {
175 return nil
176 }
177 app := string(name)
178 if ownership.App == "" {
179 ownership.App = app
180 }
181 record, err := ensureAppRecordLocked(bucket, app, true)
182 if err != nil {
183 return err
184 }
185 if ownership.ID == "" {
186 ownership.ID = record.ID
187 }
188 return putOwnershipHistoryLocked(history, ownership)
189 }); err != nil {
190 return err
191 }
192 }
193
194 if err := meta.Put(keyOwnershipMigration, []byte("1")); err != nil {
195 return fmt.Errorf("appstate: record ownership migration: %w", domain.ErrAppStateIO)
196 }
197 return nil
198}
199
200// putOwnershipHistoryLocked writes one incarnation's ownership record
201// into the history bucket. Records without an incarnation ID have no
202// stable key and are skipped.
203func putOwnershipHistoryLocked(history *bolt.Bucket, ownership domain.AppOwnership) error {
204 if history == nil || ownership.ID == "" {
205 return nil
206 }
207 raw, err := marshal(ownership)
208 if err != nil {
209 return err
210 }
211 if err := history.Put([]byte(ownership.ID), raw); err != nil {
212 return fmt.Errorf("appstate: record ownership history: %w", domain.ErrAppStateIO)
213 }
214 return nil
215}
216
217// marshal encodes a record as canonical JSON.
218func marshal(v any) ([]byte, error) {
219 raw, err := json.Marshal(v)
220 if err != nil {
221 return nil, fmt.Errorf("appstate: encode: %w", domain.ErrAppStateIO)
222 }
223 return raw, nil
224}
225
226// unmarshal decodes JSON; broken payloads map to ErrAppStateCorrupt.
227func unmarshal(raw []byte, v any, what string) error {
228 if err := json.Unmarshal(raw, v); err != nil {
229 return fmt.Errorf("appstate: corrupt %s: %w", what, domain.ErrAppStateCorrupt)
230 }
231 return nil
232}
233
234// appBucket returns the per-app bucket, creating it on writes.
235func appBucket(tx *bolt.Tx, app string, create bool) (*bolt.Bucket, error) {
236 apps := tx.Bucket(bucketApps)
237 if apps == nil {
238 return nil, fmt.Errorf("appstate: missing apps bucket: %w", domain.ErrAppStateIO)
239 }
240 if create {
241 bucket, err := apps.CreateBucketIfNotExists([]byte(app))
242 if err != nil {
243 return nil, fmt.Errorf("appstate: create app bucket: %w", domain.ErrAppStateIO)
244 }
245 return bucket, nil
246 }
247 return apps.Bucket([]byte(app)), nil
248}
249
250// subBucket returns a named child bucket, creating it on writes.
251func subBucket(parent *bolt.Bucket, name []byte, create bool) (*bolt.Bucket, error) {
252 if parent == nil {
253 return nil, nil
254 }
255 if create {
256 bucket, err := parent.CreateBucketIfNotExists(name)
257 if err != nil {
258 return nil, fmt.Errorf("appstate: create bucket %s: %w", name, domain.ErrAppStateIO)
259 }
260 return bucket, nil
261 }
262 return parent.Bucket(name), nil
263}
264
265// ensureAppRecordLocked returns the stable UUID for an app, assigning one
266// on first write. Reads leave the database untouched.
267func ensureAppRecordLocked(bucket *bolt.Bucket, app string, create bool) (appRecord, error) {
268 var record appRecord
269 raw := bucket.Get(keyAppRecord)
270 if raw == nil {
271 if !create {
272 return appRecord{Name: app}, nil
273 }
274 record = appRecord{ID: newAppID(), Name: app}
275 encoded, err := marshal(record)
276 if err != nil {
277 return appRecord{}, err
278 }
279 if err := bucket.Put(keyAppRecord, encoded); err != nil {
280 return appRecord{}, fmt.Errorf("appstate: record app id: %w", domain.ErrAppStateIO)
281 }
282 return record, nil
283 }
284 if err := unmarshal(raw, &record, "app"); err != nil {
285 return appRecord{}, err
286 }
287 if record.Name == "" {
288 record.Name = app
289 }
290 if record.ID == "" && create {
291 record.ID = newAppID()
292 encoded, err := marshal(record)
293 if err != nil {
294 return appRecord{}, err
295 }
296 if err := bucket.Put(keyAppRecord, encoded); err != nil {
297 return appRecord{}, fmt.Errorf("appstate: record app id: %w", domain.ErrAppStateIO)
298 }
299 }
300 return record, nil
301}
302
303// newAppID allocates a stable internal app UUID.
304func newAppID() string {
305 return "app-" + uuid.NewString()
306}
307
308// checkCtx aborts before opening a transaction when canceled.
309func checkCtx(ctx context.Context) error {
310 if err := ctx.Err(); err != nil {
311 return err
312 }
313 return nil
314}
315
316// getRecord reads one JSON record from a bucket.
317func getRecord(bucket *bolt.Bucket, key []byte, v any, what string) (bool, error) {
318 if bucket == nil {
319 return false, nil
320 }
321 raw := bucket.Get(key)
322 if raw == nil {
323 return false, nil
324 }
325 if err := unmarshal(raw, v, what); err != nil {
326 return false, err
327 }
328 return true, nil
329}
330
331// putRecord writes one JSON record into a bucket.
332func putRecord(bucket *bolt.Bucket, key []byte, v any) error {
333 raw, err := marshal(v)
334 if err != nil {
335 return err
336 }
337 if err := bucket.Put(key, raw); err != nil {
338 return fmt.Errorf("appstate: write record: %w", domain.ErrAppStateIO)
339 }
340 return nil
341}
342
343// listKeys returns sorted string keys of a child bucket.
344func listKeys(parent *bolt.Bucket, name []byte, reverse bool) ([]string, error) {
345 child, err := subBucket(parent, name, false)
346 if err != nil {
347 return nil, err
348 }
349 if child == nil {
350 return nil, nil
351 }
352 var keys []string
353 if err := child.ForEach(func(k, _ []byte) error {
354 keys = append(keys, string(k))
355 return nil
356 }); err != nil {
357 return nil, fmt.Errorf("appstate: list records: %w", domain.ErrAppStateIO)
358 }
359 if reverse {
360 sort.Sort(sort.Reverse(sort.StringSlice(keys)))
361 } else {
362 sort.Strings(keys)
363 }
364 return keys, nil
365}
366
367// Recover implements out.AppState.
368func (s *Store) Recover(ctx context.Context) error {
369 if err := checkCtx(ctx); err != nil {
370 return err
371 }
372 return s.db.Update(func(tx *bolt.Tx) error {
373 appsBucket := tx.Bucket(bucketApps)
374 if appsBucket == nil {
375 return nil
376 }
377 return appsBucket.ForEach(func(app, _ []byte) error {
378 // Skip nested-bucket cursor noise: app entries are buckets.
379 bucket := appsBucket.Bucket(app)
380 if bucket == nil {
381 return nil
382 }
383 return materializeCommittedLocked(tx, bucket)
384 })
385 })
386}
387
388// LoadCheckpoint implements out.AppState.
389func (s *Store) LoadCheckpoint(ctx context.Context) (domain.AppStoreCheckpoint, error) {
390 if err := checkCtx(ctx); err != nil {
391 return domain.AppStoreCheckpoint{}, err
392 }
393 var checkpoint domain.AppStoreCheckpoint
394 if err := s.db.View(func(tx *bolt.Tx) error {
395 bucket := tx.Bucket(bucketCheckpoint)
396 if bucket == nil {
397 checkpoint = domain.AppStoreCheckpoint{Version: domain.AppStoreVersion}
398 return nil
399 }
400 raw := bucket.Get(keyStoreRecord)
401 if raw == nil {
402 checkpoint = domain.AppStoreCheckpoint{Version: domain.AppStoreVersion}
403 return nil
404 }
405 return unmarshal(raw, &checkpoint, "store")
406 }); err != nil {
407 return domain.AppStoreCheckpoint{}, err
408 }
409 if checkpoint.Version > domain.AppStoreVersion {
410 return domain.AppStoreCheckpoint{}, fmt.Errorf("appstate: unsupported store version %d: %w", checkpoint.Version, domain.ErrAppStateIncompatible)
411 }
412 return checkpoint, nil
413}
414
415// ListApps implements out.AppState.
416func (s *Store) ListApps(ctx context.Context) ([]string, error) {
417 if err := checkCtx(ctx); err != nil {
418 return nil, err
419 }
420 var apps []string
421 if err := s.db.View(func(tx *bolt.Tx) error {
422 appsBucket := tx.Bucket(bucketApps)
423 if appsBucket == nil {
424 return nil
425 }
426 return appsBucket.ForEach(func(app, _ []byte) error {
427 if appsBucket.Bucket(app) != nil {
428 apps = append(apps, string(app))
429 }
430 return nil
431 })
432 }); err != nil {
433 return nil, err
434 }
435 sort.Strings(apps)
436 return apps, nil
437}
438
439// LoadDesired implements out.AppState.
440func (s *Store) LoadDesired(ctx context.Context, app string) (domain.AppDesiredRevision, bool, error) {
441 if err := checkCtx(ctx); err != nil {
442 return domain.AppDesiredRevision{}, false, err
443 }
444 var desired domain.AppDesiredRevision
445 var ok bool
446 if err := s.db.View(func(tx *bolt.Tx) error {
447 bucket, err := appBucket(tx, app, false)
448 if err != nil {
449 return err
450 }
451 found, err := getRecord(bucket, keyDesired, &desired, "desired")
452 if err != nil {
453 return err
454 }
455 ok = found
456 return nil
457 }); err != nil {
458 return domain.AppDesiredRevision{}, false, err
459 }
460 return desired, ok, nil
461}
462
463// LoadRevision implements out.AppState.
464func (s *Store) LoadRevision(ctx context.Context, app, revision string) (domain.AppDesiredRevision, error) {
465 if err := checkCtx(ctx); err != nil {
466 return domain.AppDesiredRevision{}, err
467 }
468 var rev domain.AppDesiredRevision
469 if err := s.db.View(func(tx *bolt.Tx) error {
470 bucket, err := appBucket(tx, app, false)
471 if err != nil {
472 return err
473 }
474 revisions, err := subBucket(bucket, subRevisions, false)
475 if err != nil {
476 return err
477 }
478 if revisions == nil {
479 return fmt.Errorf("appstate: revision %s not found: %w", revision, domain.ErrAppRevisionNotFound)
480 }
481 raw := revisions.Get([]byte(revision))
482 if raw == nil {
483 return fmt.Errorf("appstate: revision %s not found: %w", revision, domain.ErrAppRevisionNotFound)
484 }
485 return unmarshal(raw, &rev, "revision")
486 }); err != nil {
487 return domain.AppDesiredRevision{}, err
488 }
489 return rev, nil
490}
491
492// ListRevisions implements out.AppState.
493func (s *Store) ListRevisions(ctx context.Context, app string) ([]string, error) {
494 if err := checkCtx(ctx); err != nil {
495 return nil, err
496 }
497 var revs []string
498 if err := s.db.View(func(tx *bolt.Tx) error {
499 bucket, err := appBucket(tx, app, false)
500 if err != nil {
501 return err
502 }
503 keys, err := listKeys(bucket, subRevisions, true)
504 if err != nil {
505 return err
506 }
507 revs = keys
508 return nil
509 }); err != nil {
510 return nil, err
511 }
512 return revs, nil
513}
514
515// LoadActive implements out.AppState.
516func (s *Store) LoadActive(ctx context.Context, app string) (domain.AppActive, bool, error) {
517 if err := checkCtx(ctx); err != nil {
518 return domain.AppActive{}, false, err
519 }
520 var active domain.AppActive
521 var ok bool
522 if err := s.db.View(func(tx *bolt.Tx) error {
523 bucket, err := appBucket(tx, app, false)
524 if err != nil {
525 return err
526 }
527 found, err := getRecord(bucket, keyActive, &active, "active")
528 if err != nil {
529 return err
530 }
531 ok = found
532 return nil
533 }); err != nil {
534 return domain.AppActive{}, false, err
535 }
536 return active, ok, nil
537}
538
539// LoadIntent implements out.AppState. Absent intent means running.
540func (s *Store) LoadIntent(ctx context.Context, app string) (domain.AppStopIntent, error) {
541 if err := checkCtx(ctx); err != nil {
542 return domain.AppStopIntent{}, err
543 }
544 var intent domain.AppStopIntent
545 if err := s.db.View(func(tx *bolt.Tx) error {
546 bucket, err := appBucket(tx, app, false)
547 if err != nil {
548 return err
549 }
550 found, err := getRecord(bucket, keyIntent, &intent, "intent")
551 if err != nil {
552 return err
553 }
554 if !found {
555 intent = domain.AppStopIntent{App: app}
556 }
557 return nil
558 }); err != nil {
559 return domain.AppStopIntent{}, err
560 }
561 return intent, nil
562}
563
564// SaveIntent implements out.AppState.
565func (s *Store) SaveIntent(ctx context.Context, intent domain.AppStopIntent) error {
566 if err := checkCtx(ctx); err != nil {
567 return err
568 }
569 return s.db.Update(func(tx *bolt.Tx) error {
570 bucket, err := appBucket(tx, intent.App, true)
571 if err != nil {
572 return err
573 }
574 if _, err := ensureAppRecordLocked(bucket, intent.App, true); err != nil {
575 return err
576 }
577 return putRecord(bucket, keyIntent, intent)
578 })
579}
580
581// LoadOwnership implements out.AppState.
582func (s *Store) LoadOwnership(ctx context.Context, app string) (domain.AppOwnership, error) {
583 if err := checkCtx(ctx); err != nil {
584 return domain.AppOwnership{}, err
585 }
586 var ownership domain.AppOwnership
587 if err := s.db.View(func(tx *bolt.Tx) error {
588 bucket, err := appBucket(tx, app, false)
589 if err != nil {
590 return err
591 }
592 if bucket == nil {
593 ownership = domain.AppOwnership{App: app}
594 return nil
595 }
596 found, err := getRecord(bucket, keyOwnership, &ownership, "ownership")
597 if err != nil {
598 return err
599 }
600 if !found {
601 record, err := ensureAppRecordLocked(bucket, app, false)
602 if err != nil {
603 return err
604 }
605 ownership = domain.AppOwnership{App: app, ID: record.ID}
606 }
607 return nil
608 }); err != nil {
609 return domain.AppOwnership{}, err
610 }
611 return ownership, nil
612}
613
614// LoadRecoveryInhibitions implements out.AppState.
615func (s *Store) LoadRecoveryInhibitions(ctx context.Context, app string) ([]domain.AppRecoveryInhibition, error) {
616 if err := checkCtx(ctx); err != nil {
617 return nil, err
618 }
619 var inhibitions []domain.AppRecoveryInhibition
620 if err := s.db.View(func(tx *bolt.Tx) error {
621 bucket, err := appBucket(tx, app, false)
622 if err != nil {
623 return err
624 }
625 _, err = getRecord(bucket, keyInhibitions, &inhibitions, "recovery inhibitions")
626 return err
627 }); err != nil {
628 return nil, err
629 }
630 return inhibitions, nil
631}
632
633// SaveRecoveryInhibition implements out.AppState.
634func (s *Store) SaveRecoveryInhibition(ctx context.Context, inhibition domain.AppRecoveryInhibition) error {
635 if err := checkCtx(ctx); err != nil {
636 return err
637 }
638 return s.db.Update(func(tx *bolt.Tx) error {
639 bucket, err := appBucket(tx, inhibition.App, true)
640 if err != nil {
641 return err
642 }
643 record, err := ensureAppRecordLocked(bucket, inhibition.App, true)
644 if err != nil {
645 return err
646 }
647 if inhibition.AppID == "" {
648 inhibition.AppID = record.ID
649 }
650 var inhibitions []domain.AppRecoveryInhibition
651 if found, err := getRecord(bucket, keyInhibitions, &inhibitions, "recovery inhibitions"); err != nil {
652 return err
653 } else if !found {
654 inhibitions = nil
655 }
656 replaced := false
657 for i, existing := range inhibitions {
658 if existing.Service == inhibition.Service && existing.ContainerID == inhibition.ContainerID {
659 inhibitions[i] = inhibition
660 replaced = true
661 break
662 }
663 }
664 if !replaced {
665 inhibitions = append(inhibitions, inhibition)
666 }
667 return putRecord(bucket, keyInhibitions, inhibitions)
668 })
669}
670
671// ClearRecoveryInhibition implements out.AppState.
672func (s *Store) ClearRecoveryInhibition(ctx context.Context, app, service, containerID string) error {
673 if err := checkCtx(ctx); err != nil {
674 return err
675 }
676 return s.db.Update(func(tx *bolt.Tx) error {
677 bucket, err := appBucket(tx, app, false)
678 if err != nil || bucket == nil {
679 return err
680 }
681 var inhibitions []domain.AppRecoveryInhibition
682 found, err := getRecord(bucket, keyInhibitions, &inhibitions, "recovery inhibitions")
683 if err != nil || !found {
684 return err
685 }
686 kept := make([]domain.AppRecoveryInhibition, 0, len(inhibitions))
687 for _, existing := range inhibitions {
688 if existing.Service == service && existing.ContainerID == containerID {
689 continue
690 }
691 kept = append(kept, existing)
692 }
693 if len(kept) == len(inhibitions) {
694 return nil
695 }
696 return putRecord(bucket, keyInhibitions, kept)
697 })
698}
699
700// SaveOwnership implements out.AppState.
701func (s *Store) SaveOwnership(ctx context.Context, ownership domain.AppOwnership) error {
702 if err := checkCtx(ctx); err != nil {
703 return err
704 }
705 return s.db.Update(func(tx *bolt.Tx) error {
706 bucket, err := appBucket(tx, ownership.App, true)
707 if err != nil {
708 return err
709 }
710 record, err := ensureAppRecordLocked(bucket, ownership.App, true)
711 if err != nil {
712 return err
713 }
714 if ownership.ID == "" {
715 ownership.ID = record.ID
716 }
717 // A new UUID means a new incarnation: never carry the old
718 // incarnation's resources implicitly. The old incarnation's
719 // history record stays in the ownership-history bucket.
720 if ownership.ID != record.ID && len(ownership.Volumes) == 0 && len(ownership.Secrets) == 0 && len(ownership.Networks) == 0 {
721 record = appRecord{ID: ownership.ID, Name: ownership.App}
722 encoded, err := marshal(record)
723 if err != nil {
724 return err
725 }
726 if err := bucket.Put(keyAppRecord, encoded); err != nil {
727 return fmt.Errorf("appstate: record app id: %w", domain.ErrAppStateIO)
728 }
729 }
730 if err := putRecord(bucket, keyOwnership, ownership); err != nil {
731 return err
732 }
733 return putOwnershipHistoryLocked(tx.Bucket(bucketOwnershipHistory), ownership)
734 })
735}
736
737// StageApply implements out.AppState.
738func (s *Store) StageApply(ctx context.Context, intent domain.AppApplyIntent) error {
739 if err := checkCtx(ctx); err != nil {
740 return err
741 }
742 intent.State = domain.AppIntentStaged
743 return s.db.Update(func(tx *bolt.Tx) error {
744 bucket, err := appBucket(tx, intent.App, true)
745 if err != nil {
746 return err
747 }
748 if _, err := ensureAppRecordLocked(bucket, intent.App, true); err != nil {
749 return err
750 }
751 intents, err := subBucket(bucket, subIntents, true)
752 if err != nil {
753 return err
754 }
755 return putRecord(intents, []byte(intent.Intent), intent)
756 })
757}
758
759// CommitApply implements out.AppState: the single atomic commit point.
760func (s *Store) CommitApply(ctx context.Context, app, intentID string) error {
761 if err := checkCtx(ctx); err != nil {
762 return err
763 }
764 return s.db.Update(func(tx *bolt.Tx) error {
765 bucket, err := appBucket(tx, app, true)
766 if err != nil {
767 return err
768 }
769 intent, err := loadApplyIntentLocked(bucket, app, intentID)
770 if err != nil {
771 return err
772 }
773 if intent.State != domain.AppIntentStaged {
774 return fmt.Errorf("appstate: intent %s is %s, not staged: %w", intentID, intent.State, domain.ErrAppStateConflict)
775 }
776 if err := verifySupersedesLocked(bucket, intent); err != nil {
777 return err
778 }
779 intent.State = domain.AppIntentCommitted
780 intents, err := subBucket(bucket, subIntents, true)
781 if err != nil {
782 return err
783 }
784 return putRecord(intents, []byte(intentID), intent)
785 })
786}
787
788// MaterializeApply implements out.AppState. Idempotent: safe to replay.
789func (s *Store) MaterializeApply(ctx context.Context, app, intentID string) error {
790 if err := checkCtx(ctx); err != nil {
791 return err
792 }
793 return s.db.Update(func(tx *bolt.Tx) error {
794 bucket, err := appBucket(tx, app, true)
795 if err != nil {
796 return err
797 }
798 intent, err := loadApplyIntentLocked(bucket, app, intentID)
799 if err != nil {
800 return err
801 }
802 if intent.State == domain.AppIntentStaged {
803 return fmt.Errorf("appstate: intent %s is staged, commit first: %w", intentID, domain.ErrAppStateConflict)
804 }
805 if intent.State == domain.AppIntentApplied {
806 return nil
807 }
808 if err := materializeIntentLocked(tx, bucket, intent); err != nil {
809 return err
810 }
811 intent.State = domain.AppIntentApplied
812 intents, err := subBucket(bucket, subIntents, true)
813 if err != nil {
814 return err
815 }
816 return putRecord(intents, []byte(intentID), intent)
817 })
818}
819
820// materializeIntentLocked writes revision + desired + checkpoint deltas.
821func materializeIntentLocked(tx *bolt.Tx, bucket *bolt.Bucket, intent domain.AppApplyIntent) error {
822 if err := verifySupersedesLocked(bucket, intent); err != nil {
823 return err
824 }
825 rev := domain.AppDesiredRevision{
826 Revision: intent.Revision,
827 App: intent.App,
828 Supersedes: intent.Supersedes,
829 AcceptedAt: intent.CreatedAt,
830 SourceSHA256: intent.SourceSHA256,
831 Spec: intent.Spec,
832 Reservations: intent.Reservations,
833 SecretsRequired: secretsRequired(intent.Spec),
834 SecretsEnv: secretsEnv(intent.Spec),
835 Status: domain.AppStepPending,
836 }
837 revisions, err := subBucket(bucket, subRevisions, true)
838 if err != nil {
839 return err
840 }
841 if err := putRecord(revisions, []byte(intent.Revision), rev); err != nil {
842 return err
843 }
844 if err := putRecord(bucket, keyDesired, rev); err != nil {
845 return err
846 }
847 return foldReservationsLocked(tx, intent)
848}
849
850// foldReservationsLocked folds one committed intent's deltas into the checkpoint.
851func foldReservationsLocked(tx *bolt.Tx, intent domain.AppApplyIntent) error {
852 bucket := tx.Bucket(bucketCheckpoint)
853 if bucket == nil {
854 return fmt.Errorf("appstate: missing checkpoint bucket: %w", domain.ErrAppStateIO)
855 }
856 var checkpoint domain.AppStoreCheckpoint
857 raw := bucket.Get(keyStoreRecord)
858 if raw == nil {
859 checkpoint = domain.AppStoreCheckpoint{Version: domain.AppStoreVersion}
860 } else if err := unmarshal(raw, &checkpoint, "store"); err != nil {
861 return err
862 }
863 kept := checkpoint.Reservations[:0]
864 for _, res := range checkpoint.Reservations {
865 if res.App != intent.App {
866 kept = append(kept, res)
867 }
868 }
869 kept = append(kept, intent.Reservations...)
870 checkpoint.Version = domain.AppStoreVersion
871 checkpoint.Reservations = kept
872 encoded, err := marshal(checkpoint)
873 if err != nil {
874 return err
875 }
876 if err := bucket.Put(keyStoreRecord, encoded); err != nil {
877 return fmt.Errorf("appstate: write checkpoint: %w", domain.ErrAppStateIO)
878 }
879 return nil
880}
881
882// materializeCommittedLocked finishes all committed intents for one app.
883// The caller holds the write tx, so revision, desired, checkpoint, and
884// intent updates commit atomically.
885func materializeCommittedLocked(tx *bolt.Tx, bucket *bolt.Bucket) error {
886 intents, err := subBucket(bucket, subIntents, false)
887 if err != nil {
888 return err
889 }
890 if intents == nil {
891 return nil
892 }
893 type pending struct {
894 id string
895 intent domain.AppApplyIntent
896 }
897 var committed []pending
898 if err := intents.ForEach(func(k, v []byte) error {
899 var intent domain.AppApplyIntent
900 if err := unmarshal(v, &intent, "intent"); err != nil {
901 return err
902 }
903 if intent.State == domain.AppIntentCommitted {
904 committed = append(committed, pending{id: string(k), intent: intent})
905 }
906 return nil
907 }); err != nil {
908 return err
909 }
910 for _, item := range committed {
911 if err := materializeIntentLocked(tx, bucket, item.intent); err != nil {
912 if errors.Is(err, domain.ErrAppStateConflict) {
913 // A newer apply superseded this committed intent while the
914 // daemon was away; it must never overwrite newer desired
915 // state and can be dropped.
916 if delErr := intents.Delete([]byte(item.id)); delErr != nil {
917 return fmt.Errorf("appstate: drop superseded intent: %w", domain.ErrAppStateIO)
918 }
919 continue
920 }
921 return err
922 }
923 item.intent.State = domain.AppIntentApplied
924 if err := putRecord(intents, []byte(item.id), item.intent); err != nil {
925 return err
926 }
927 }
928 return nil
929}
930
931// verifySupersedesLocked rejects an intent whose expected former revision
932// no longer matches the current desired revision: a concurrent apply that
933// committed first wins, and the stale intent can never overwrite it.
934func verifySupersedesLocked(bucket *bolt.Bucket, intent domain.AppApplyIntent) error {
935 var desired domain.AppDesiredRevision
936 found, err := getRecord(bucket, keyDesired, &desired, "desired")
937 if err != nil {
938 return err
939 }
940 current := ""
941 if found {
942 current = desired.Revision
943 }
944 if current != intent.Supersedes {
945 return fmt.Errorf("appstate: intent %s expected desired %q but found %q: %w",
946 intent.Intent, intent.Supersedes, current, domain.ErrAppStateConflict)
947 }
948 return nil
949}
950
951// LoadApplyIntent implements out.AppState.
952func (s *Store) LoadApplyIntent(ctx context.Context, app, intentID string) (domain.AppApplyIntent, error) {
953 if err := checkCtx(ctx); err != nil {
954 return domain.AppApplyIntent{}, err
955 }
956 var intent domain.AppApplyIntent
957 if err := s.db.View(func(tx *bolt.Tx) error {
958 bucket, err := appBucket(tx, app, false)
959 if err != nil {
960 return err
961 }
962 loaded, err := loadApplyIntentLocked(bucket, app, intentID)
963 if err != nil {
964 return err
965 }
966 intent = loaded
967 return nil
968 }); err != nil {
969 return domain.AppApplyIntent{}, err
970 }
971 return intent, nil
972}
973
974func loadApplyIntentLocked(bucket *bolt.Bucket, app, intentID string) (domain.AppApplyIntent, error) {
975 var intent domain.AppApplyIntent
976 if bucket == nil {
977 return domain.AppApplyIntent{}, fmt.Errorf("appstate: intent %s not found: %w", intentID, domain.ErrAppIntentNotFound)
978 }
979 intents, err := subBucket(bucket, subIntents, false)
980 if err != nil {
981 return domain.AppApplyIntent{}, err
982 }
983 if intents == nil {
984 return domain.AppApplyIntent{}, fmt.Errorf("appstate: intent %s not found: %w", intentID, domain.ErrAppIntentNotFound)
985 }
986 raw := intents.Get([]byte(intentID))
987 if raw == nil {
988 return domain.AppApplyIntent{}, fmt.Errorf("appstate: intent %s not found: %w", intentID, domain.ErrAppIntentNotFound)
989 }
990 if err := unmarshal(raw, &intent, "intent"); err != nil {
991 return domain.AppApplyIntent{}, err
992 }
993 _ = app
994 return intent, nil
995}
996
997// ListIntents implements out.AppState.
998func (s *Store) ListIntents(ctx context.Context, app string) ([]string, error) {
999 if err := checkCtx(ctx); err != nil {
1000 return nil, err
1001 }
1002 var ids []string
1003 if err := s.db.View(func(tx *bolt.Tx) error {
1004 bucket, err := appBucket(tx, app, false)
1005 if err != nil {
1006 return err
1007 }
1008 keys, err := listKeys(bucket, subIntents, false)
1009 if err != nil {
1010 return err
1011 }
1012 ids = keys
1013 return nil
1014 }); err != nil {
1015 return nil, err
1016 }
1017 return ids, nil
1018}
1019
1020// CollectGarbage implements out.AppState.
1021func (s *Store) CollectGarbage(ctx context.Context, app string, inFlight []string) error {
1022 if err := checkCtx(ctx); err != nil {
1023 return err
1024 }
1025 return s.db.Update(func(tx *bolt.Tx) error {
1026 bucket, err := appBucket(tx, app, true)
1027 if err != nil {
1028 return err
1029 }
1030 protected, err := protectedRevisionsLocked(bucket, app, inFlight)
1031 if err != nil {
1032 return err
1033 }
1034 if err := sweepIntentsLocked(bucket, inFlight); err != nil {
1035 return err
1036 }
1037 return sweepRevisionsLocked(bucket, protected, s.Retention())
1038 })
1039}
1040
1041// protectedRevisionsLocked returns revisions that GC must never evict.
1042func protectedRevisionsLocked(bucket *bolt.Bucket, app string, inFlight []string) (map[string]struct{}, error) {
1043 protected := map[string]struct{}{}
1044 for _, rev := range inFlight {
1045 protected[rev] = struct{}{}
1046 }
1047 var desired domain.AppDesiredRevision
1048 if found, err := getRecord(bucket, keyDesired, &desired, "desired"); err != nil {
1049 return nil, err
1050 } else if found {
1051 protected[desired.Revision] = struct{}{}
1052 }
1053 var active domain.AppActive
1054 if found, err := getRecord(bucket, keyActive, &active, "active"); err != nil {
1055 return nil, err
1056 } else if found {
1057 for _, svc := range active.Services {
1058 protected[svc.EffectiveRevision] = struct{}{}
1059 }
1060 }
1061 intents, err := subBucket(bucket, subIntents, false)
1062 if err != nil {
1063 return nil, err
1064 }
1065 if intents != nil {
1066 if err := intents.ForEach(func(_, v []byte) error {
1067 var intent domain.AppApplyIntent
1068 if err := unmarshal(v, &intent, "intent"); err != nil {
1069 return err
1070 }
1071 // Only unfinished work pins its revision: staged orphans
1072 // and committed-but-unmaterialized intents. Applied intents
1073 // are history; their revisions age out by retention.
1074 if intent.State == domain.AppIntentStaged || intent.State == domain.AppIntentCommitted {
1075 protected[intent.Revision] = struct{}{}
1076 }
1077 return nil
1078 }); err != nil {
1079 return nil, err
1080 }
1081 }
1082 _ = app
1083 return protected, nil
1084}
1085
1086// sweepIntentsLocked removes applied intents and orphaned staged intents.
1087// Committed intents are never swept here: they carry unmaterialized
1088// deltas that recovery must complete.
1089func sweepIntentsLocked(bucket *bolt.Bucket, inFlight []string) error {
1090 intents, err := subBucket(bucket, subIntents, false)
1091 if err != nil {
1092 return err
1093 }
1094 if intents == nil {
1095 return nil
1096 }
1097 // A live apply's intent must never be swept as an orphan: the caller
1098 // names its own in-flight intent so concurrent work is preserved.
1099 protected := make(map[string]struct{}, len(inFlight))
1100 for _, id := range inFlight {
1101 protected[id] = struct{}{}
1102 }
1103 var drop [][]byte
1104 if err := intents.ForEach(func(k, v []byte) error {
1105 if _, ok := protected[string(k)]; ok {
1106 return nil
1107 }
1108 var intent domain.AppApplyIntent
1109 if err := unmarshal(v, &intent, "intent"); err != nil {
1110 return err
1111 }
1112 if intent.State == domain.AppIntentApplied || intent.State == domain.AppIntentStaged {
1113 key := append([]byte(nil), k...)
1114 drop = append(drop, key)
1115 }
1116 return nil
1117 }); err != nil {
1118 return err
1119 }
1120 for _, key := range drop {
1121 if err := intents.Delete(key); err != nil {
1122 return fmt.Errorf("appstate: remove intent: %w", domain.ErrAppStateIO)
1123 }
1124 }
1125 return nil
1126}
1127
1128// sweepRevisionsLocked removes unreferenced revisions beyond retention.
1129func sweepRevisionsLocked(bucket *bolt.Bucket, protected map[string]struct{}, retention int) error {
1130 revisions, err := subBucket(bucket, subRevisions, false)
1131 if err != nil {
1132 return err
1133 }
1134 if revisions == nil {
1135 return nil
1136 }
1137 var keys []string
1138 if err := revisions.ForEach(func(k, _ []byte) error {
1139 keys = append(keys, string(k))
1140 return nil
1141 }); err != nil {
1142 return fmt.Errorf("appstate: list revisions: %w", domain.ErrAppStateIO)
1143 }
1144 sort.Sort(sort.Reverse(sort.StringSlice(keys)))
1145 kept := 0
1146 for _, rev := range keys {
1147 if _, ok := protected[rev]; ok {
1148 continue
1149 }
1150 if kept >= retention {
1151 if err := revisions.Delete([]byte(rev)); err != nil {
1152 return fmt.Errorf("appstate: remove revision: %w", domain.ErrAppStateIO)
1153 }
1154 continue
1155 }
1156 kept++
1157 }
1158 return nil
1159}
1160
1161// SaveOperation implements out.AppState.
1162func (s *Store) SaveOperation(ctx context.Context, op domain.AppOperation) error {
1163 if err := checkCtx(ctx); err != nil {
1164 return err
1165 }
1166 return s.db.Update(func(tx *bolt.Tx) error {
1167 bucket, err := appBucket(tx, op.App, true)
1168 if err != nil {
1169 return err
1170 }
1171 if _, err := ensureAppRecordLocked(bucket, op.App, true); err != nil {
1172 return err
1173 }
1174 ops, err := subBucket(bucket, subOps, true)
1175 if err != nil {
1176 return err
1177 }
1178 return putRecord(ops, []byte(op.Op), op)
1179 })
1180}
1181
1182// AppExists implements out.AppState: a read-only live-identity check
1183// that never creates an app bucket.
1184func (s *Store) AppExists(ctx context.Context, app string) (bool, error) {
1185 if err := checkCtx(ctx); err != nil {
1186 return false, err
1187 }
1188 exists := false
1189 err := s.db.View(func(tx *bolt.Tx) error {
1190 bucket, err := appBucket(tx, app, false)
1191 if err != nil {
1192 return err
1193 }
1194 if bucket == nil {
1195 return nil
1196 }
1197 known, err := appHasLiveIdentityLocked(bucket)
1198 if err != nil {
1199 return err
1200 }
1201 exists = known
1202 return nil
1203 })
1204 if err != nil {
1205 return false, err
1206 }
1207 return exists, nil
1208}
1209
1210// ClaimOperation implements out.AppState: one atomic check-and-write for
1211// a mutation request key.
1212func (s *Store) ClaimOperation(ctx context.Context, candidate domain.AppOperation) (domain.AppOperation, bool, error) {
1213 if err := checkCtx(ctx); err != nil {
1214 return domain.AppOperation{}, false, err
1215 }
1216 if candidate.Op == "" {
1217 return domain.AppOperation{}, false, fmt.Errorf("appstate: claim operation without a key: %w", domain.ErrInvalidAppSpec)
1218 }
1219 var existing domain.AppOperation
1220 claimed := false
1221 err := s.db.Update(func(tx *bolt.Tx) error {
1222 bucket, err := appBucket(tx, candidate.App, false)
1223 if err != nil {
1224 return err
1225 }
1226 if stored, ok, err := claimedOperation(bucket, candidate); err != nil {
1227 return err
1228 } else if ok {
1229 existing = stored
1230 return nil
1231 }
1232 if bucket == nil {
1233 return fmt.Errorf("appstate: app %q has no state: %w", candidate.App, domain.ErrAppNotFound)
1234 }
1235 known, err := appHasLiveIdentityLocked(bucket)
1236 if err != nil {
1237 return err
1238 }
1239 if !known {
1240 return fmt.Errorf("appstate: app %q has no live identity: %w", candidate.App, domain.ErrAppNotFound)
1241 }
1242 writeBucket, err := appBucket(tx, candidate.App, true)
1243 if err != nil {
1244 return err
1245 }
1246 if _, err := ensureAppRecordLocked(writeBucket, candidate.App, true); err != nil {
1247 return err
1248 }
1249 ops, err := subBucket(writeBucket, subOps, true)
1250 if err != nil {
1251 return err
1252 }
1253 if err := putRecord(ops, []byte(candidate.Op), candidate); err != nil {
1254 return err
1255 }
1256 existing = candidate
1257 claimed = true
1258 return nil
1259 })
1260 if err != nil {
1261 return domain.AppOperation{}, false, err
1262 }
1263 return existing, claimed, nil
1264}
1265
1266// claimedOperation returns the journal already stored under the candidate
1267// key when its request identity matches. A key answering a different
1268// request identity is a conflict, never a replay.
1269func claimedOperation(bucket *bolt.Bucket, candidate domain.AppOperation) (domain.AppOperation, bool, error) {
1270 if bucket == nil {
1271 return domain.AppOperation{}, false, nil
1272 }
1273 ops, err := subBucket(bucket, subOps, false)
1274 if err != nil || ops == nil {
1275 return domain.AppOperation{}, false, err
1276 }
1277 raw := ops.Get([]byte(candidate.Op))
1278 if raw == nil {
1279 return domain.AppOperation{}, false, nil
1280 }
1281 var stored domain.AppOperation
1282 if err := unmarshal(raw, &stored, "operation"); err != nil {
1283 return domain.AppOperation{}, false, err
1284 }
1285 if stored.Request != candidate.Request {
1286 return domain.AppOperation{}, false, fmt.Errorf(
1287 "appstate: operation key %s already answered a different request: %w",
1288 candidate.Op, domain.ErrAppStateConflict,
1289 )
1290 }
1291 return stored, true, nil
1292}
1293
1294// appHasLiveIdentityLocked reports whether an existing app bucket holds a
1295// live identity: desired state, active state, or an ownership
1296// incarnation. Operation journals alone are not identity — a retired app
1297// keeps them so its own key still replays, while a new key must never
1298// recreate state for the freed name.
1299func appHasLiveIdentityLocked(bucket *bolt.Bucket) (bool, error) {
1300 if bucket.Get(keyDesired) != nil || bucket.Get(keyActive) != nil {
1301 return true, nil
1302 }
1303 var ownership domain.AppOwnership
1304 found, err := getRecord(bucket, keyOwnership, &ownership, "ownership")
1305 if err != nil {
1306 return false, err
1307 }
1308 return found && ownership.ID != "", nil
1309}
1310
1311// LoadOperation implements out.AppState.
1312func (s *Store) LoadOperation(ctx context.Context, app, opID string) (domain.AppOperation, error) {
1313 if err := checkCtx(ctx); err != nil {
1314 return domain.AppOperation{}, err
1315 }
1316 var op domain.AppOperation
1317 if err := s.db.View(func(tx *bolt.Tx) error {
1318 bucket, err := appBucket(tx, app, false)
1319 if err != nil {
1320 return err
1321 }
1322 ops, err := subBucket(bucket, subOps, false)
1323 if err != nil {
1324 return err
1325 }
1326 if ops == nil {
1327 return fmt.Errorf("appstate: operation %s not found: %w", opID, domain.ErrAppOperationNotFound)
1328 }
1329 raw := ops.Get([]byte(opID))
1330 if raw == nil {
1331 return fmt.Errorf("appstate: operation %s not found: %w", opID, domain.ErrAppOperationNotFound)
1332 }
1333 return unmarshal(raw, &op, "operation")
1334 }); err != nil {
1335 return domain.AppOperation{}, err
1336 }
1337 return op, nil
1338}
1339
1340// LoadLatestOperation implements out.AppState.
1341func (s *Store) LoadLatestOperation(ctx context.Context, app string) (domain.AppOperation, bool, error) {
1342 if err := checkCtx(ctx); err != nil {
1343 return domain.AppOperation{}, false, err
1344 }
1345 var latest domain.AppOperation
1346 found := false
1347 err := s.db.View(func(tx *bolt.Tx) error {
1348 bucket, err := appBucket(tx, app, false)
1349 if err != nil {
1350 return err
1351 }
1352 ops, err := subBucket(bucket, subOps, false)
1353 if err != nil || ops == nil {
1354 return err
1355 }
1356 return ops.ForEach(func(_, raw []byte) error {
1357 var op domain.AppOperation
1358 if err := unmarshal(raw, &op, "operation"); err != nil {
1359 return err
1360 }
1361 if !found || op.StartedAt.After(latest.StartedAt) {
1362 latest, found = op, true
1363 }
1364 return nil
1365 })
1366 })
1367 if err != nil {
1368 return domain.AppOperation{}, false, err
1369 }
1370 return latest, found, nil
1371}
1372
1373// RetireApp implements out.AppState. In one transaction it archives the
1374// live incarnation's ownership record (with every resource marked
1375// retained), drops the live ownership record, resets the app's UUID so a
1376// later reapply allocates a NEW incarnation, and clears desired, active,
1377// intent, staged intents, and revisions. The archived record stays the
1378// positive proof that retained volumes and secrets exist, so prune keeps
1379// protecting them while a name reuse can never adopt them.
1380func (s *Store) RetireApp(ctx context.Context, app string) error {
1381 if err := checkCtx(ctx); err != nil {
1382 return err
1383 }
1384 return s.db.Update(func(tx *bolt.Tx) error {
1385 bucket, err := appBucket(tx, app, true)
1386 if err != nil {
1387 return err
1388 }
1389 if err := archiveRetainedOwnershipLocked(tx, bucket); err != nil {
1390 return err
1391 }
1392 return clearIncarnationStateLocked(bucket, app)
1393 })
1394}
1395
1396// archiveRetainedOwnershipLocked archives the live ownership record with
1397// every resource marked retained, so prune keeps protecting them. A
1398// missing record is not an error.
1399func archiveRetainedOwnershipLocked(tx *bolt.Tx, bucket *bolt.Bucket) error {
1400 var ownership domain.AppOwnership
1401 found, err := getRecord(bucket, keyOwnership, &ownership, "ownership")
1402 if err != nil {
1403 return err
1404 }
1405 if !found || ownership.ID == "" {
1406 return nil
1407 }
1408 for i := range ownership.Volumes {
1409 ownership.Volumes[i].State = domain.AppResourceRetained
1410 }
1411 for i := range ownership.Secrets {
1412 ownership.Secrets[i].State = domain.AppResourceRetained
1413 }
1414 for i := range ownership.Images {
1415 ownership.Images[i].State = domain.AppResourceReleased
1416 }
1417 return putOwnershipHistoryLocked(tx.Bucket(bucketOwnershipHistory), ownership)
1418}
1419
1420// clearIncarnationStateLocked drops the live ownership record and all
1421// operational state, then resets the app record so the next write assigns
1422// a fresh incarnation UUID.
1423func clearIncarnationStateLocked(bucket *bolt.Bucket, app string) error {
1424 for _, key := range [][]byte{keyOwnership, keyDesired, keyActive, keyIntent, keyInhibitions} {
1425 if err := bucket.Delete(key); err != nil {
1426 return fmt.Errorf("appstate: retire %s: %w", app, domain.ErrAppStateIO)
1427 }
1428 }
1429 for _, sub := range [][]byte{subRevisions, subIntents} {
1430 if err := bucket.DeleteBucket(sub); err != nil && !errors.Is(err, bolterrors.ErrBucketNotFound) {
1431 return fmt.Errorf("appstate: retire %s: %w", app, domain.ErrAppStateIO)
1432 }
1433 }
1434 encoded, err := marshal(appRecord{Name: app})
1435 if err != nil {
1436 return err
1437 }
1438 if err := bucket.Put(keyAppRecord, encoded); err != nil {
1439 return fmt.Errorf("appstate: reset app record: %w", domain.ErrAppStateIO)
1440 }
1441 return nil
1442}
1443
1444// SaveActive implements out.AppState.
1445func (s *Store) SaveActive(ctx context.Context, active domain.AppActive) error {
1446 if err := checkCtx(ctx); err != nil {
1447 return err
1448 }
1449 return s.db.Update(func(tx *bolt.Tx) error {
1450 bucket, err := appBucket(tx, active.App, true)
1451 if err != nil {
1452 return err
1453 }
1454 if _, err := ensureAppRecordLocked(bucket, active.App, true); err != nil {
1455 return err
1456 }
1457 return putRecord(bucket, keyActive, active)
1458 })
1459}
1460
1461// RegisterBackendBinds implements out.AppState. Backend claims join the
1462// global checkpoint in the same write tx that checks them, so two
1463// concurrent deploys cannot both claim one bind.
1464func (s *Store) RegisterBackendBinds(ctx context.Context, binds []domain.AppListenerReservation) error {
1465 if err := checkCtx(ctx); err != nil {
1466 return err
1467 }
1468 if err := validateBackendClaims(binds); err != nil {
1469 return err
1470 }
1471 return s.db.Update(func(tx *bolt.Tx) error {
1472 checkpoint, bucket, err := loadCheckpointLocked(tx)
1473 if err != nil {
1474 return err
1475 }
1476 if err := checkBackendConflicts(checkpoint.Reservations, binds); err != nil {
1477 return err
1478 }
1479 checkpoint.Reservations = replaceContainerClaims(checkpoint.Reservations, binds)
1480 checkpoint.Version = domain.AppStoreVersion
1481 return putCheckpointLocked(bucket, checkpoint)
1482 })
1483}
1484
1485// validateBackendClaims rejects non-backend or malformed claims before
1486// any storage mutation.
1487func validateBackendClaims(binds []domain.AppListenerReservation) error {
1488 for _, bind := range binds {
1489 if bind.Owner != domain.OwnerGordonBackend {
1490 return fmt.Errorf("appstate: backend claim owner must be %q: %w", domain.OwnerGordonBackend, domain.ErrAppReservationConflict)
1491 }
1492 if bind.Proto != "tcp" && bind.Proto != "udp" {
1493 return fmt.Errorf("appstate: backend claim proto must be tcp or udp: %w", domain.ErrAppReservationConflict)
1494 }
1495 if bind.IP != "127.0.0.1" || bind.Port <= 0 || bind.ContainerID == "" {
1496 return fmt.Errorf("appstate: malformed backend claim: %w", domain.ErrAppReservationConflict)
1497 }
1498 }
1499 return nil
1500}
1501
1502// loadCheckpointLocked reads the global checkpoint in a write tx.
1503func loadCheckpointLocked(tx *bolt.Tx) (domain.AppStoreCheckpoint, *bolt.Bucket, error) {
1504 bucket := tx.Bucket(bucketCheckpoint)
1505 if bucket == nil {
1506 return domain.AppStoreCheckpoint{}, nil, fmt.Errorf("appstate: missing checkpoint bucket: %w", domain.ErrAppStateIO)
1507 }
1508 var checkpoint domain.AppStoreCheckpoint
1509 raw := bucket.Get(keyStoreRecord)
1510 if raw == nil {
1511 return domain.AppStoreCheckpoint{Version: domain.AppStoreVersion}, bucket, nil
1512 }
1513 if err := unmarshal(raw, &checkpoint, "store"); err != nil {
1514 return domain.AppStoreCheckpoint{}, nil, err
1515 }
1516 return checkpoint, bucket, nil
1517}
1518
1519// putCheckpointLocked persists the global checkpoint in a write tx.
1520func putCheckpointLocked(bucket *bolt.Bucket, checkpoint domain.AppStoreCheckpoint) error {
1521 encoded, err := marshal(checkpoint)
1522 if err != nil {
1523 return err
1524 }
1525 if err := bucket.Put(keyStoreRecord, encoded); err != nil {
1526 return fmt.Errorf("appstate: write checkpoint: %w", domain.ErrAppStateIO)
1527 }
1528 return nil
1529}
1530
1531// checkBackendConflicts fails closed on another app's identical loopback
1532// claim. Same-app claims never self-conflict.
1533func checkBackendConflicts(existing []domain.AppListenerReservation, binds []domain.AppListenerReservation) error {
1534 for _, bind := range binds {
1535 for _, other := range existing {
1536 if other.App == bind.App {
1537 continue
1538 }
1539 if backendOverlaps(other, bind) {
1540 return fmt.Errorf("appstate: backend bind %s conflicts with %s/%s: %w",
1541 backendDescribe(bind), other.App, other.Service, domain.ErrAppReservationConflict)
1542 }
1543 }
1544 }
1545 return nil
1546}
1547
1548// replaceContainerClaims swaps one container's backend claims for the
1549// new set, preserving every other claim (old and replacement coexist
1550// until the old container's verified withdrawal).
1551func replaceContainerClaims(existing []domain.AppListenerReservation, binds []domain.AppListenerReservation) []domain.AppListenerReservation {
1552 if len(binds) == 0 {
1553 return existing
1554 }
1555 kept := existing[:0]
1556 for _, res := range existing {
1557 if res.Owner == domain.OwnerGordonBackend && res.App == binds[0].App && res.ContainerID == binds[0].ContainerID {
1558 continue
1559 }
1560 kept = append(kept, res)
1561 }
1562 return append(kept, binds...)
1563}
1564
1565// ReleaseBackendBinds implements out.AppState.
1566func (s *Store) ReleaseBackendBinds(ctx context.Context, app, containerID string) error {
1567 if err := checkCtx(ctx); err != nil {
1568 return err
1569 }
1570 return s.db.Update(func(tx *bolt.Tx) error {
1571 bucket := tx.Bucket(bucketCheckpoint)
1572 if bucket == nil {
1573 return fmt.Errorf("appstate: missing checkpoint bucket: %w", domain.ErrAppStateIO)
1574 }
1575 var checkpoint domain.AppStoreCheckpoint
1576 raw := bucket.Get(keyStoreRecord)
1577 if raw == nil {
1578 return nil
1579 }
1580 if err := unmarshal(raw, &checkpoint, "store"); err != nil {
1581 return err
1582 }
1583 kept := checkpoint.Reservations[:0]
1584 for _, existing := range checkpoint.Reservations {
1585 if existing.Owner == domain.OwnerGordonBackend && existing.App == app && existing.ContainerID == containerID {
1586 continue
1587 }
1588 kept = append(kept, existing)
1589 }
1590 checkpoint.Reservations = kept
1591 encoded, err := marshal(checkpoint)
1592 if err != nil {
1593 return err
1594 }
1595 if err := bucket.Put(keyStoreRecord, encoded); err != nil {
1596 return fmt.Errorf("appstate: write checkpoint: %w", domain.ErrAppStateIO)
1597 }
1598 return nil
1599 })
1600}
1601
1602// backendOverlaps reports exact loopback bind collisions per protocol.
1603// A TCP and a UDP claim on the same host port never collide: they are
1604// distinct sockets. Claims are exact 127.0.0.1:port allocations
1605// (kernel-arbitrated), so only the same IP+port collides — never user
1606// wildcard publishes on other IPs.
1607func backendOverlaps(a, b domain.AppListenerReservation) bool {
1608 if a.Proto != b.Proto || (a.Proto != "tcp" && a.Proto != "udp") {
1609 return false
1610 }
1611 return a.IP == "127.0.0.1" && b.IP == "127.0.0.1" && a.Port == b.Port
1612}
1613
1614func backendDescribe(bind domain.AppListenerReservation) string {
1615 return fmt.Sprintf("%s://%s:%d", bind.Proto, bind.IP, bind.Port)
1616}
1617func secretsRequired(spec domain.AppSpec) []string {
1618 var paths []string
1619 for _, svc := range spec.Services {
1620 for _, name := range svc.Secrets {
1621 paths = append(paths, domain.AppSecretPath(spec.Name, svc.Name, name))
1622 }
1623 }
1624 sort.Strings(paths)
1625 return paths
1626}
1627
1628// secretsEnv maps pass paths back to env keys.
1629func secretsEnv(spec domain.AppSpec) map[string]string {
1630 env := map[string]string{}
1631 for _, svc := range spec.Services {
1632 for envKey, name := range svc.Secrets {
1633 env[domain.AppSecretPath(spec.Name, svc.Name, name)] = envKey
1634 }
1635 }
1636 return env
1637}