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 returns revision ids ordered newest first.
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// AcceptApply implements out.AppState: stage, commit, materialize, then
738// best-effort garbage collection. Each step is its own transaction so a
739// crash leaves a state Recover or GC can finish.
740func (s *Store) AcceptApply(ctx context.Context, intent domain.AppApplyIntent) error {
741 if err := s.StageApply(ctx, intent); err != nil {
742 return fmt.Errorf("appstate: stage apply: %w", err)
743 }
744 if err := s.CommitApply(ctx, intent.App, intent.Intent); err != nil {
745 return fmt.Errorf("appstate: commit apply: %w", err)
746 }
747 if err := s.MaterializeApply(ctx, intent.App, intent.Intent); err != nil {
748 return fmt.Errorf("appstate: materialize apply (re-query by intent): %w", err)
749 }
750 if err := s.CollectGarbage(ctx, intent.App, []string{intent.Intent}); err != nil {
751 s.log.Warn().Err(err).Str("app", intent.App).Msg("appstate: garbage collection failed, continuing")
752 }
753 return nil
754}
755
756// StageApply writes a staged intent. No reservation or revision is
757// visible until CommitApply. It is one step of AcceptApply, exposed for
758// crash-recovery tests.
759func (s *Store) StageApply(ctx context.Context, intent domain.AppApplyIntent) error {
760 if err := checkCtx(ctx); err != nil {
761 return err
762 }
763 intent.State = domain.AppIntentStaged
764 return s.db.Update(func(tx *bolt.Tx) error {
765 bucket, err := appBucket(tx, intent.App, true)
766 if err != nil {
767 return err
768 }
769 if _, err := ensureAppRecordLocked(bucket, intent.App, true); err != nil {
770 return err
771 }
772 intents, err := subBucket(bucket, subIntents, true)
773 if err != nil {
774 return err
775 }
776 return putRecord(intents, []byte(intent.Intent), intent)
777 })
778}
779
780// CommitApply is the single atomic commit point: staged→committed. It is
781// one step of AcceptApply, exposed for crash-recovery tests.
782func (s *Store) CommitApply(ctx context.Context, app, intentID string) error {
783 if err := checkCtx(ctx); err != nil {
784 return err
785 }
786 return s.db.Update(func(tx *bolt.Tx) error {
787 bucket, err := appBucket(tx, app, true)
788 if err != nil {
789 return err
790 }
791 intent, err := loadApplyIntentLocked(bucket, app, intentID)
792 if err != nil {
793 return err
794 }
795 if intent.State != domain.AppIntentStaged {
796 return fmt.Errorf("appstate: intent %s is %s, not staged: %w", intentID, intent.State, domain.ErrAppStateConflict)
797 }
798 if err := verifySupersedesLocked(bucket, intent); err != nil {
799 return err
800 }
801 intent.State = domain.AppIntentCommitted
802 intents, err := subBucket(bucket, subIntents, true)
803 if err != nil {
804 return err
805 }
806 return putRecord(intents, []byte(intentID), intent)
807 })
808}
809
810// MaterializeApply finishes a committed intent. Idempotent: safe to
811// replay. It is one step of AcceptApply, exposed for crash-recovery tests.
812func (s *Store) MaterializeApply(ctx context.Context, app, intentID string) error {
813 if err := checkCtx(ctx); err != nil {
814 return err
815 }
816 return s.db.Update(func(tx *bolt.Tx) error {
817 bucket, err := appBucket(tx, app, true)
818 if err != nil {
819 return err
820 }
821 intent, err := loadApplyIntentLocked(bucket, app, intentID)
822 if err != nil {
823 return err
824 }
825 if intent.State == domain.AppIntentStaged {
826 return fmt.Errorf("appstate: intent %s is staged, commit first: %w", intentID, domain.ErrAppStateConflict)
827 }
828 if intent.State == domain.AppIntentApplied {
829 return nil
830 }
831 if err := materializeIntentLocked(tx, bucket, intent); err != nil {
832 return err
833 }
834 intent.State = domain.AppIntentApplied
835 intents, err := subBucket(bucket, subIntents, true)
836 if err != nil {
837 return err
838 }
839 return putRecord(intents, []byte(intentID), intent)
840 })
841}
842
843// materializeIntentLocked writes revision + desired + checkpoint deltas.
844func materializeIntentLocked(tx *bolt.Tx, bucket *bolt.Bucket, intent domain.AppApplyIntent) error {
845 if err := verifySupersedesLocked(bucket, intent); err != nil {
846 return err
847 }
848 rev := domain.AppDesiredRevision{
849 Revision: intent.Revision,
850 App: intent.App,
851 Supersedes: intent.Supersedes,
852 AcceptedAt: intent.CreatedAt,
853 SourceSHA256: intent.SourceSHA256,
854 Spec: intent.Spec,
855 Reservations: intent.Reservations,
856 SecretsRequired: secretsRequired(intent.Spec),
857 SecretsEnv: secretsEnv(intent.Spec),
858 Status: domain.AppStepPending,
859 }
860 revisions, err := subBucket(bucket, subRevisions, true)
861 if err != nil {
862 return err
863 }
864 if err := putRecord(revisions, []byte(intent.Revision), rev); err != nil {
865 return err
866 }
867 if err := putRecord(bucket, keyDesired, rev); err != nil {
868 return err
869 }
870 return foldReservationsLocked(tx, intent)
871}
872
873// foldReservationsLocked folds one committed intent's deltas into the checkpoint.
874func foldReservationsLocked(tx *bolt.Tx, intent domain.AppApplyIntent) error {
875 bucket := tx.Bucket(bucketCheckpoint)
876 if bucket == nil {
877 return fmt.Errorf("appstate: missing checkpoint bucket: %w", domain.ErrAppStateIO)
878 }
879 var checkpoint domain.AppStoreCheckpoint
880 raw := bucket.Get(keyStoreRecord)
881 if raw == nil {
882 checkpoint = domain.AppStoreCheckpoint{Version: domain.AppStoreVersion}
883 } else if err := unmarshal(raw, &checkpoint, "store"); err != nil {
884 return err
885 }
886 kept := checkpoint.Reservations[:0]
887 for _, res := range checkpoint.Reservations {
888 if res.App != intent.App {
889 kept = append(kept, res)
890 }
891 }
892 kept = append(kept, intent.Reservations...)
893 checkpoint.Version = domain.AppStoreVersion
894 checkpoint.Reservations = kept
895 encoded, err := marshal(checkpoint)
896 if err != nil {
897 return err
898 }
899 if err := bucket.Put(keyStoreRecord, encoded); err != nil {
900 return fmt.Errorf("appstate: write checkpoint: %w", domain.ErrAppStateIO)
901 }
902 return nil
903}
904
905// materializeCommittedLocked finishes all committed intents for one app.
906// The caller holds the write tx, so revision, desired, checkpoint, and
907// intent updates commit atomically.
908func materializeCommittedLocked(tx *bolt.Tx, bucket *bolt.Bucket) error {
909 intents, err := subBucket(bucket, subIntents, false)
910 if err != nil {
911 return err
912 }
913 if intents == nil {
914 return nil
915 }
916 type pending struct {
917 id string
918 intent domain.AppApplyIntent
919 }
920 var committed []pending
921 if err := intents.ForEach(func(k, v []byte) error {
922 var intent domain.AppApplyIntent
923 if err := unmarshal(v, &intent, "intent"); err != nil {
924 return err
925 }
926 if intent.State == domain.AppIntentCommitted {
927 committed = append(committed, pending{id: string(k), intent: intent})
928 }
929 return nil
930 }); err != nil {
931 return err
932 }
933 for _, item := range committed {
934 if err := materializeIntentLocked(tx, bucket, item.intent); err != nil {
935 if errors.Is(err, domain.ErrAppStateConflict) {
936 // A newer apply superseded this committed intent while the
937 // daemon was away; it must never overwrite newer desired
938 // state and can be dropped.
939 if delErr := intents.Delete([]byte(item.id)); delErr != nil {
940 return fmt.Errorf("appstate: drop superseded intent: %w", domain.ErrAppStateIO)
941 }
942 continue
943 }
944 return err
945 }
946 item.intent.State = domain.AppIntentApplied
947 if err := putRecord(intents, []byte(item.id), item.intent); err != nil {
948 return err
949 }
950 }
951 return nil
952}
953
954// verifySupersedesLocked rejects an intent whose expected former revision
955// no longer matches the current desired revision: a concurrent apply that
956// committed first wins, and the stale intent can never overwrite it.
957func verifySupersedesLocked(bucket *bolt.Bucket, intent domain.AppApplyIntent) error {
958 var desired domain.AppDesiredRevision
959 found, err := getRecord(bucket, keyDesired, &desired, "desired")
960 if err != nil {
961 return err
962 }
963 current := ""
964 if found {
965 current = desired.Revision
966 }
967 if current != intent.Supersedes {
968 return fmt.Errorf("appstate: intent %s expected desired %q but found %q: %w",
969 intent.Intent, intent.Supersedes, current, domain.ErrAppStateConflict)
970 }
971 return nil
972}
973
974// LoadApplyIntent returns one apply intent by id.
975func (s *Store) LoadApplyIntent(ctx context.Context, app, intentID string) (domain.AppApplyIntent, error) {
976 if err := checkCtx(ctx); err != nil {
977 return domain.AppApplyIntent{}, err
978 }
979 var intent domain.AppApplyIntent
980 if err := s.db.View(func(tx *bolt.Tx) error {
981 bucket, err := appBucket(tx, app, false)
982 if err != nil {
983 return err
984 }
985 loaded, err := loadApplyIntentLocked(bucket, app, intentID)
986 if err != nil {
987 return err
988 }
989 intent = loaded
990 return nil
991 }); err != nil {
992 return domain.AppApplyIntent{}, err
993 }
994 return intent, nil
995}
996
997func loadApplyIntentLocked(bucket *bolt.Bucket, app, intentID string) (domain.AppApplyIntent, error) {
998 var intent domain.AppApplyIntent
999 if bucket == nil {
1000 return domain.AppApplyIntent{}, fmt.Errorf("appstate: intent %s not found: %w", intentID, domain.ErrAppIntentNotFound)
1001 }
1002 intents, err := subBucket(bucket, subIntents, false)
1003 if err != nil {
1004 return domain.AppApplyIntent{}, err
1005 }
1006 if intents == nil {
1007 return domain.AppApplyIntent{}, fmt.Errorf("appstate: intent %s not found: %w", intentID, domain.ErrAppIntentNotFound)
1008 }
1009 raw := intents.Get([]byte(intentID))
1010 if raw == nil {
1011 return domain.AppApplyIntent{}, fmt.Errorf("appstate: intent %s not found: %w", intentID, domain.ErrAppIntentNotFound)
1012 }
1013 if err := unmarshal(raw, &intent, "intent"); err != nil {
1014 return domain.AppApplyIntent{}, err
1015 }
1016 _ = app
1017 return intent, nil
1018}
1019
1020// ListIntents returns apply intent ids for recovery scans.
1021func (s *Store) ListIntents(ctx context.Context, app string) ([]string, error) {
1022 if err := checkCtx(ctx); err != nil {
1023 return nil, err
1024 }
1025 var ids []string
1026 if err := s.db.View(func(tx *bolt.Tx) error {
1027 bucket, err := appBucket(tx, app, false)
1028 if err != nil {
1029 return err
1030 }
1031 keys, err := listKeys(bucket, subIntents, false)
1032 if err != nil {
1033 return err
1034 }
1035 ids = keys
1036 return nil
1037 }); err != nil {
1038 return nil, err
1039 }
1040 return ids, nil
1041}
1042
1043// CollectGarbage removes staged orphans, applied intents, and
1044// unreferenced revisions beyond retention. Referenced revisions
1045// (desired/active/in-flight) are never evicted.
1046func (s *Store) CollectGarbage(ctx context.Context, app string, inFlight []string) error {
1047 if err := checkCtx(ctx); err != nil {
1048 return err
1049 }
1050 return s.db.Update(func(tx *bolt.Tx) error {
1051 bucket, err := appBucket(tx, app, true)
1052 if err != nil {
1053 return err
1054 }
1055 protected, err := protectedRevisionsLocked(bucket, app, inFlight)
1056 if err != nil {
1057 return err
1058 }
1059 if err := sweepIntentsLocked(bucket, inFlight); err != nil {
1060 return err
1061 }
1062 return sweepRevisionsLocked(bucket, protected, s.Retention())
1063 })
1064}
1065
1066// protectedRevisionsLocked returns revisions that GC must never evict.
1067func protectedRevisionsLocked(bucket *bolt.Bucket, app string, inFlight []string) (map[string]struct{}, error) {
1068 protected := map[string]struct{}{}
1069 for _, rev := range inFlight {
1070 protected[rev] = struct{}{}
1071 }
1072 var desired domain.AppDesiredRevision
1073 if found, err := getRecord(bucket, keyDesired, &desired, "desired"); err != nil {
1074 return nil, err
1075 } else if found {
1076 protected[desired.Revision] = struct{}{}
1077 }
1078 var active domain.AppActive
1079 if found, err := getRecord(bucket, keyActive, &active, "active"); err != nil {
1080 return nil, err
1081 } else if found {
1082 for _, svc := range active.Services {
1083 protected[svc.EffectiveRevision] = struct{}{}
1084 }
1085 }
1086 intents, err := subBucket(bucket, subIntents, false)
1087 if err != nil {
1088 return nil, err
1089 }
1090 if intents != nil {
1091 if err := intents.ForEach(func(_, v []byte) error {
1092 var intent domain.AppApplyIntent
1093 if err := unmarshal(v, &intent, "intent"); err != nil {
1094 return err
1095 }
1096 // Only unfinished work pins its revision: staged orphans
1097 // and committed-but-unmaterialized intents. Applied intents
1098 // are history; their revisions age out by retention.
1099 if intent.State == domain.AppIntentStaged || intent.State == domain.AppIntentCommitted {
1100 protected[intent.Revision] = struct{}{}
1101 }
1102 return nil
1103 }); err != nil {
1104 return nil, err
1105 }
1106 }
1107 _ = app
1108 return protected, nil
1109}
1110
1111// sweepIntentsLocked removes applied intents and orphaned staged intents.
1112// Committed intents are never swept here: they carry unmaterialized
1113// deltas that recovery must complete.
1114func sweepIntentsLocked(bucket *bolt.Bucket, inFlight []string) error {
1115 intents, err := subBucket(bucket, subIntents, false)
1116 if err != nil {
1117 return err
1118 }
1119 if intents == nil {
1120 return nil
1121 }
1122 // A live apply's intent must never be swept as an orphan: the caller
1123 // names its own in-flight intent so concurrent work is preserved.
1124 protected := make(map[string]struct{}, len(inFlight))
1125 for _, id := range inFlight {
1126 protected[id] = struct{}{}
1127 }
1128 var drop [][]byte
1129 if err := intents.ForEach(func(k, v []byte) error {
1130 if _, ok := protected[string(k)]; ok {
1131 return nil
1132 }
1133 var intent domain.AppApplyIntent
1134 if err := unmarshal(v, &intent, "intent"); err != nil {
1135 return err
1136 }
1137 if intent.State == domain.AppIntentApplied || intent.State == domain.AppIntentStaged {
1138 key := append([]byte(nil), k...)
1139 drop = append(drop, key)
1140 }
1141 return nil
1142 }); err != nil {
1143 return err
1144 }
1145 for _, key := range drop {
1146 if err := intents.Delete(key); err != nil {
1147 return fmt.Errorf("appstate: remove intent: %w", domain.ErrAppStateIO)
1148 }
1149 }
1150 return nil
1151}
1152
1153// sweepRevisionsLocked removes unreferenced revisions beyond retention.
1154func sweepRevisionsLocked(bucket *bolt.Bucket, protected map[string]struct{}, retention int) error {
1155 revisions, err := subBucket(bucket, subRevisions, false)
1156 if err != nil {
1157 return err
1158 }
1159 if revisions == nil {
1160 return nil
1161 }
1162 var keys []string
1163 if err := revisions.ForEach(func(k, _ []byte) error {
1164 keys = append(keys, string(k))
1165 return nil
1166 }); err != nil {
1167 return fmt.Errorf("appstate: list revisions: %w", domain.ErrAppStateIO)
1168 }
1169 sort.Sort(sort.Reverse(sort.StringSlice(keys)))
1170 kept := 0
1171 for _, rev := range keys {
1172 if _, ok := protected[rev]; ok {
1173 continue
1174 }
1175 if kept >= retention {
1176 if err := revisions.Delete([]byte(rev)); err != nil {
1177 return fmt.Errorf("appstate: remove revision: %w", domain.ErrAppStateIO)
1178 }
1179 continue
1180 }
1181 kept++
1182 }
1183 return nil
1184}
1185
1186// SaveOperation implements out.AppState.
1187func (s *Store) SaveOperation(ctx context.Context, op domain.AppOperation) error {
1188 if err := checkCtx(ctx); err != nil {
1189 return err
1190 }
1191 return s.db.Update(func(tx *bolt.Tx) error {
1192 bucket, err := appBucket(tx, op.App, true)
1193 if err != nil {
1194 return err
1195 }
1196 if _, err := ensureAppRecordLocked(bucket, op.App, true); err != nil {
1197 return err
1198 }
1199 ops, err := subBucket(bucket, subOps, true)
1200 if err != nil {
1201 return err
1202 }
1203 return putRecord(ops, []byte(op.Op), op)
1204 })
1205}
1206
1207// AppExists implements out.AppState: a read-only live-identity check
1208// that never creates an app bucket.
1209func (s *Store) AppExists(ctx context.Context, app string) (bool, error) {
1210 if err := checkCtx(ctx); err != nil {
1211 return false, err
1212 }
1213 exists := false
1214 err := s.db.View(func(tx *bolt.Tx) error {
1215 bucket, err := appBucket(tx, app, false)
1216 if err != nil {
1217 return err
1218 }
1219 if bucket == nil {
1220 return nil
1221 }
1222 known, err := appHasLiveIdentityLocked(bucket)
1223 if err != nil {
1224 return err
1225 }
1226 exists = known
1227 return nil
1228 })
1229 if err != nil {
1230 return false, err
1231 }
1232 return exists, nil
1233}
1234
1235// ClaimOperation implements out.AppState: one atomic check-and-write for
1236// a mutation request key.
1237func (s *Store) ClaimOperation(ctx context.Context, candidate domain.AppOperation) (domain.AppOperation, bool, error) {
1238 if err := checkCtx(ctx); err != nil {
1239 return domain.AppOperation{}, false, err
1240 }
1241 if candidate.Op == "" {
1242 return domain.AppOperation{}, false, fmt.Errorf("appstate: claim operation without a key: %w", domain.ErrInvalidAppSpec)
1243 }
1244 var existing domain.AppOperation
1245 claimed := false
1246 err := s.db.Update(func(tx *bolt.Tx) error {
1247 bucket, err := appBucket(tx, candidate.App, false)
1248 if err != nil {
1249 return err
1250 }
1251 if stored, ok, err := claimedOperation(bucket, candidate); err != nil {
1252 return err
1253 } else if ok {
1254 existing = stored
1255 return nil
1256 }
1257 if bucket == nil {
1258 return fmt.Errorf("appstate: app %q has no state: %w", candidate.App, domain.ErrAppNotFound)
1259 }
1260 known, err := appHasLiveIdentityLocked(bucket)
1261 if err != nil {
1262 return err
1263 }
1264 if !known {
1265 return fmt.Errorf("appstate: app %q has no live identity: %w", candidate.App, domain.ErrAppNotFound)
1266 }
1267 writeBucket, err := appBucket(tx, candidate.App, true)
1268 if err != nil {
1269 return err
1270 }
1271 if _, err := ensureAppRecordLocked(writeBucket, candidate.App, true); err != nil {
1272 return err
1273 }
1274 ops, err := subBucket(writeBucket, subOps, true)
1275 if err != nil {
1276 return err
1277 }
1278 if err := putRecord(ops, []byte(candidate.Op), candidate); err != nil {
1279 return err
1280 }
1281 existing = candidate
1282 claimed = true
1283 return nil
1284 })
1285 if err != nil {
1286 return domain.AppOperation{}, false, err
1287 }
1288 return existing, claimed, nil
1289}
1290
1291// claimedOperation returns the journal already stored under the candidate
1292// key when its request identity matches. A key answering a different
1293// request identity is a conflict, never a replay.
1294func claimedOperation(bucket *bolt.Bucket, candidate domain.AppOperation) (domain.AppOperation, bool, error) {
1295 if bucket == nil {
1296 return domain.AppOperation{}, false, nil
1297 }
1298 ops, err := subBucket(bucket, subOps, false)
1299 if err != nil || ops == nil {
1300 return domain.AppOperation{}, false, err
1301 }
1302 raw := ops.Get([]byte(candidate.Op))
1303 if raw == nil {
1304 return domain.AppOperation{}, false, nil
1305 }
1306 var stored domain.AppOperation
1307 if err := unmarshal(raw, &stored, "operation"); err != nil {
1308 return domain.AppOperation{}, false, err
1309 }
1310 if stored.Request != candidate.Request {
1311 return domain.AppOperation{}, false, fmt.Errorf(
1312 "appstate: operation key %s already answered a different request: %w",
1313 candidate.Op, domain.ErrAppStateConflict,
1314 )
1315 }
1316 return stored, true, nil
1317}
1318
1319// appHasLiveIdentityLocked reports whether an existing app bucket holds a
1320// live identity: desired state, active state, or an ownership
1321// incarnation. Operation journals alone are not identity — a retired app
1322// keeps them so its own key still replays, while a new key must never
1323// recreate state for the freed name.
1324func appHasLiveIdentityLocked(bucket *bolt.Bucket) (bool, error) {
1325 if bucket.Get(keyDesired) != nil || bucket.Get(keyActive) != nil {
1326 return true, nil
1327 }
1328 var ownership domain.AppOwnership
1329 found, err := getRecord(bucket, keyOwnership, &ownership, "ownership")
1330 if err != nil {
1331 return false, err
1332 }
1333 return found && ownership.ID != "", nil
1334}
1335
1336// LoadOperation implements out.AppState.
1337func (s *Store) LoadOperation(ctx context.Context, app, opID string) (domain.AppOperation, error) {
1338 if err := checkCtx(ctx); err != nil {
1339 return domain.AppOperation{}, err
1340 }
1341 var op domain.AppOperation
1342 if err := s.db.View(func(tx *bolt.Tx) error {
1343 bucket, err := appBucket(tx, app, false)
1344 if err != nil {
1345 return err
1346 }
1347 ops, err := subBucket(bucket, subOps, false)
1348 if err != nil {
1349 return err
1350 }
1351 if ops == nil {
1352 return fmt.Errorf("appstate: operation %s not found: %w", opID, domain.ErrAppOperationNotFound)
1353 }
1354 raw := ops.Get([]byte(opID))
1355 if raw == nil {
1356 return fmt.Errorf("appstate: operation %s not found: %w", opID, domain.ErrAppOperationNotFound)
1357 }
1358 return unmarshal(raw, &op, "operation")
1359 }); err != nil {
1360 return domain.AppOperation{}, err
1361 }
1362 return op, nil
1363}
1364
1365// LoadLatestOperation implements out.AppState.
1366func (s *Store) LoadLatestOperation(ctx context.Context, app string) (domain.AppOperation, bool, error) {
1367 if err := checkCtx(ctx); err != nil {
1368 return domain.AppOperation{}, false, err
1369 }
1370 var latest domain.AppOperation
1371 found := false
1372 err := s.db.View(func(tx *bolt.Tx) error {
1373 bucket, err := appBucket(tx, app, false)
1374 if err != nil {
1375 return err
1376 }
1377 ops, err := subBucket(bucket, subOps, false)
1378 if err != nil || ops == nil {
1379 return err
1380 }
1381 return ops.ForEach(func(_, raw []byte) error {
1382 var op domain.AppOperation
1383 if err := unmarshal(raw, &op, "operation"); err != nil {
1384 return err
1385 }
1386 if !found || op.StartedAt.After(latest.StartedAt) {
1387 latest, found = op, true
1388 }
1389 return nil
1390 })
1391 })
1392 if err != nil {
1393 return domain.AppOperation{}, false, err
1394 }
1395 return latest, found, nil
1396}
1397
1398// RetireApp implements out.AppState. In one transaction it archives the
1399// live incarnation's ownership record (with every resource marked
1400// retained), drops the live ownership record, resets the app's UUID so a
1401// later reapply allocates a NEW incarnation, and clears desired, active,
1402// intent, staged intents, and revisions. The archived record stays the
1403// positive proof that retained volumes and secrets exist, so prune keeps
1404// protecting them while a name reuse can never adopt them.
1405func (s *Store) RetireApp(ctx context.Context, app string) error {
1406 if err := checkCtx(ctx); err != nil {
1407 return err
1408 }
1409 return s.db.Update(func(tx *bolt.Tx) error {
1410 bucket, err := appBucket(tx, app, true)
1411 if err != nil {
1412 return err
1413 }
1414 if err := archiveRetainedOwnershipLocked(tx, bucket); err != nil {
1415 return err
1416 }
1417 return clearIncarnationStateLocked(bucket, app)
1418 })
1419}
1420
1421// archiveRetainedOwnershipLocked archives the live ownership record with
1422// every resource marked retained, so prune keeps protecting them. A
1423// missing record is not an error.
1424func archiveRetainedOwnershipLocked(tx *bolt.Tx, bucket *bolt.Bucket) error {
1425 var ownership domain.AppOwnership
1426 found, err := getRecord(bucket, keyOwnership, &ownership, "ownership")
1427 if err != nil {
1428 return err
1429 }
1430 if !found || ownership.ID == "" {
1431 return nil
1432 }
1433 for i := range ownership.Volumes {
1434 ownership.Volumes[i].State = domain.AppResourceRetained
1435 }
1436 for i := range ownership.Secrets {
1437 ownership.Secrets[i].State = domain.AppResourceRetained
1438 }
1439 for i := range ownership.Images {
1440 ownership.Images[i].State = domain.AppResourceReleased
1441 }
1442 return putOwnershipHistoryLocked(tx.Bucket(bucketOwnershipHistory), ownership)
1443}
1444
1445// clearIncarnationStateLocked drops the live ownership record and all
1446// operational state, then resets the app record so the next write assigns
1447// a fresh incarnation UUID.
1448func clearIncarnationStateLocked(bucket *bolt.Bucket, app string) error {
1449 for _, key := range [][]byte{keyOwnership, keyDesired, keyActive, keyIntent, keyInhibitions} {
1450 if err := bucket.Delete(key); err != nil {
1451 return fmt.Errorf("appstate: retire %s: %w", app, domain.ErrAppStateIO)
1452 }
1453 }
1454 for _, sub := range [][]byte{subRevisions, subIntents} {
1455 if err := bucket.DeleteBucket(sub); err != nil && !errors.Is(err, bolterrors.ErrBucketNotFound) {
1456 return fmt.Errorf("appstate: retire %s: %w", app, domain.ErrAppStateIO)
1457 }
1458 }
1459 encoded, err := marshal(appRecord{Name: app})
1460 if err != nil {
1461 return err
1462 }
1463 if err := bucket.Put(keyAppRecord, encoded); err != nil {
1464 return fmt.Errorf("appstate: reset app record: %w", domain.ErrAppStateIO)
1465 }
1466 return nil
1467}
1468
1469// SaveActive implements out.AppState.
1470func (s *Store) SaveActive(ctx context.Context, active domain.AppActive) error {
1471 if err := checkCtx(ctx); err != nil {
1472 return err
1473 }
1474 return s.db.Update(func(tx *bolt.Tx) error {
1475 bucket, err := appBucket(tx, active.App, true)
1476 if err != nil {
1477 return err
1478 }
1479 if _, err := ensureAppRecordLocked(bucket, active.App, true); err != nil {
1480 return err
1481 }
1482 return putRecord(bucket, keyActive, active)
1483 })
1484}
1485
1486// RegisterBackendBinds implements out.AppState. Backend claims join the
1487// global checkpoint in the same write tx that checks them, so two
1488// concurrent deploys cannot both claim one bind.
1489func (s *Store) RegisterBackendBinds(ctx context.Context, binds []domain.AppListenerReservation) error {
1490 if err := checkCtx(ctx); err != nil {
1491 return err
1492 }
1493 if err := validateBackendClaims(binds); err != nil {
1494 return err
1495 }
1496 return s.db.Update(func(tx *bolt.Tx) error {
1497 checkpoint, bucket, err := loadCheckpointLocked(tx)
1498 if err != nil {
1499 return err
1500 }
1501 if err := checkBackendConflicts(checkpoint.Reservations, binds); err != nil {
1502 return err
1503 }
1504 checkpoint.Reservations = replaceContainerClaims(checkpoint.Reservations, binds)
1505 checkpoint.Version = domain.AppStoreVersion
1506 return putCheckpointLocked(bucket, checkpoint)
1507 })
1508}
1509
1510// validateBackendClaims rejects non-backend or malformed claims before
1511// any storage mutation.
1512func validateBackendClaims(binds []domain.AppListenerReservation) error {
1513 for _, bind := range binds {
1514 if bind.Owner != domain.OwnerGordonBackend {
1515 return fmt.Errorf("appstate: backend claim owner must be %q: %w", domain.OwnerGordonBackend, domain.ErrAppReservationConflict)
1516 }
1517 if bind.Proto != "tcp" && bind.Proto != "udp" {
1518 return fmt.Errorf("appstate: backend claim proto must be tcp or udp: %w", domain.ErrAppReservationConflict)
1519 }
1520 if bind.IP != "127.0.0.1" || bind.Port <= 0 || bind.ContainerID == "" {
1521 return fmt.Errorf("appstate: malformed backend claim: %w", domain.ErrAppReservationConflict)
1522 }
1523 }
1524 return nil
1525}
1526
1527// loadCheckpointLocked reads the global checkpoint in a write tx.
1528func loadCheckpointLocked(tx *bolt.Tx) (domain.AppStoreCheckpoint, *bolt.Bucket, error) {
1529 bucket := tx.Bucket(bucketCheckpoint)
1530 if bucket == nil {
1531 return domain.AppStoreCheckpoint{}, nil, fmt.Errorf("appstate: missing checkpoint bucket: %w", domain.ErrAppStateIO)
1532 }
1533 var checkpoint domain.AppStoreCheckpoint
1534 raw := bucket.Get(keyStoreRecord)
1535 if raw == nil {
1536 return domain.AppStoreCheckpoint{Version: domain.AppStoreVersion}, bucket, nil
1537 }
1538 if err := unmarshal(raw, &checkpoint, "store"); err != nil {
1539 return domain.AppStoreCheckpoint{}, nil, err
1540 }
1541 return checkpoint, bucket, nil
1542}
1543
1544// putCheckpointLocked persists the global checkpoint in a write tx.
1545func putCheckpointLocked(bucket *bolt.Bucket, checkpoint domain.AppStoreCheckpoint) error {
1546 encoded, err := marshal(checkpoint)
1547 if err != nil {
1548 return err
1549 }
1550 if err := bucket.Put(keyStoreRecord, encoded); err != nil {
1551 return fmt.Errorf("appstate: write checkpoint: %w", domain.ErrAppStateIO)
1552 }
1553 return nil
1554}
1555
1556// checkBackendConflicts fails closed on another app's identical loopback
1557// claim. Same-app claims never self-conflict.
1558func checkBackendConflicts(existing []domain.AppListenerReservation, binds []domain.AppListenerReservation) error {
1559 for _, bind := range binds {
1560 for _, other := range existing {
1561 if other.App == bind.App {
1562 continue
1563 }
1564 if backendOverlaps(other, bind) {
1565 return fmt.Errorf("appstate: backend bind %s conflicts with %s/%s: %w",
1566 backendDescribe(bind), other.App, other.Service, domain.ErrAppReservationConflict)
1567 }
1568 }
1569 }
1570 return nil
1571}
1572
1573// replaceContainerClaims swaps one container's backend claims for the
1574// new set, preserving every other claim (old and replacement coexist
1575// until the old container's verified withdrawal).
1576func replaceContainerClaims(existing []domain.AppListenerReservation, binds []domain.AppListenerReservation) []domain.AppListenerReservation {
1577 if len(binds) == 0 {
1578 return existing
1579 }
1580 kept := existing[:0]
1581 for _, res := range existing {
1582 if res.Owner == domain.OwnerGordonBackend && res.App == binds[0].App && res.ContainerID == binds[0].ContainerID {
1583 continue
1584 }
1585 kept = append(kept, res)
1586 }
1587 return append(kept, binds...)
1588}
1589
1590// ReleaseBackendBinds implements out.AppState.
1591func (s *Store) ReleaseBackendBinds(ctx context.Context, app, containerID string) error {
1592 if err := checkCtx(ctx); err != nil {
1593 return err
1594 }
1595 return s.db.Update(func(tx *bolt.Tx) error {
1596 bucket := tx.Bucket(bucketCheckpoint)
1597 if bucket == nil {
1598 return fmt.Errorf("appstate: missing checkpoint bucket: %w", domain.ErrAppStateIO)
1599 }
1600 var checkpoint domain.AppStoreCheckpoint
1601 raw := bucket.Get(keyStoreRecord)
1602 if raw == nil {
1603 return nil
1604 }
1605 if err := unmarshal(raw, &checkpoint, "store"); err != nil {
1606 return err
1607 }
1608 kept := checkpoint.Reservations[:0]
1609 for _, existing := range checkpoint.Reservations {
1610 if existing.Owner == domain.OwnerGordonBackend && existing.App == app && existing.ContainerID == containerID {
1611 continue
1612 }
1613 kept = append(kept, existing)
1614 }
1615 checkpoint.Reservations = kept
1616 encoded, err := marshal(checkpoint)
1617 if err != nil {
1618 return err
1619 }
1620 if err := bucket.Put(keyStoreRecord, encoded); err != nil {
1621 return fmt.Errorf("appstate: write checkpoint: %w", domain.ErrAppStateIO)
1622 }
1623 return nil
1624 })
1625}
1626
1627// backendOverlaps reports exact loopback bind collisions per protocol.
1628// A TCP and a UDP claim on the same host port never collide: they are
1629// distinct sockets. Claims are exact 127.0.0.1:port allocations
1630// (kernel-arbitrated), so only the same IP+port collides — never user
1631// wildcard publishes on other IPs.
1632func backendOverlaps(a, b domain.AppListenerReservation) bool {
1633 if a.Proto != b.Proto || (a.Proto != "tcp" && a.Proto != "udp") {
1634 return false
1635 }
1636 return a.IP == "127.0.0.1" && b.IP == "127.0.0.1" && a.Port == b.Port
1637}
1638
1639func backendDescribe(bind domain.AppListenerReservation) string {
1640 return fmt.Sprintf("%s://%s:%d", bind.Proto, bind.IP, bind.Port)
1641}
1642func secretsRequired(spec domain.AppSpec) []string {
1643 var paths []string
1644 for _, svc := range spec.Services {
1645 for _, name := range svc.Secrets {
1646 paths = append(paths, domain.AppSecretPath(spec.Name, svc.Name, name))
1647 }
1648 }
1649 sort.Strings(paths)
1650 return paths
1651}
1652
1653// secretsEnv maps pass paths back to env keys.
1654func secretsEnv(spec domain.AppSpec) map[string]string {
1655 env := map[string]string{}
1656 for _, svc := range spec.Services {
1657 for envKey, name := range svc.Secrets {
1658 env[domain.AppSecretPath(spec.Name, svc.Name, name)] = envKey
1659 }
1660 }
1661 return env
1662}