| 1 | package host |
| 2 | |
| 3 | import ( |
| 4 | "context" |
| 5 | "errors" |
| 6 | "os" |
| 7 | "strings" |
| 8 | |
| 9 | "github.com/carlosframework/platform/internal/hostbox" |
| 10 | ) |
| 11 | |
| 12 | // Called after identity-checked event removal with namespace, slot and record |
| 13 | // locks held. Latest slot references remain replayable; legacy consumption is |
| 14 | // separate authority and is never collected here. |
| 15 | func (l *servingParkLedger) collectClosedLocked(ctx context.Context, r *servingParkRecord, slot servingParkSlot) error { |
| 16 | if err := ctx.Err(); err != nil { |
| 17 | return err |
| 18 | } |
| 19 | if !r.Complete || !r.Closed || !r.EventRemoved { |
| 20 | return nil |
| 21 | } |
| 22 | if r.CollectAfter.IsZero() { |
| 23 | // Old records start their window when first observed fully complete. |
| 24 | // Clock rollback must not move the deadline backwards on a later pass. |
| 25 | now := l.now().UTC() |
| 26 | if now.Before(r.ParkedAt) { |
| 27 | return nil |
| 28 | } |
| 29 | r.CollectAfter = now.Add(hostbox.JournalReceiptRetention) |
| 30 | return l.save(*r) |
| 31 | } |
| 32 | if l.now().Before(r.CollectAfter) || slot.ID == r.Authorization.ID || slot.Pending == r.Authorization.ID || slot.Last == r.Authorization.ID { |
| 33 | return nil |
| 34 | } |
| 35 | if slot.Last == "" { |
| 36 | return hostbox.ErrChanged |
| 37 | } |
| 38 | if r.Authorization.Legacy { |
| 39 | consumed, err := l.legacyConsumption(r.Authorization.Successor) |
| 40 | if err != nil || consumed == nil || !sameServingLegacyConsumption(*consumed, *r) { |
| 41 | return hostbox.ErrChanged |
| 42 | } |
| 43 | // Distinguish collection from a crash after consumption was recorded |
| 44 | // but before the first retirement intent was saved. |
| 45 | if !consumed.Collected { |
| 46 | consumed.Collected = true |
| 47 | historical := servingParkRecord{Authorization: consumed.Authorization, Source: consumed.Source} |
| 48 | if err := l.bindRecord(&historical, true); err != nil { |
| 49 | return err |
| 50 | } |
| 51 | consumed.Authorization, consumed.Source = historical.Authorization, historical.Source |
| 52 | if err := l.write(servingLegacyKey("legacy", r.Authorization.Successor), consumed); err != nil { |
| 53 | return err |
| 54 | } |
| 55 | } |
| 56 | } |
| 57 | // Preparations have no authority without their retirement. Remove them |
| 58 | // while that record still makes interrupted collection discoverable. |
| 59 | names, err := l.state.Names() |
| 60 | if err != nil { |
| 61 | return err |
| 62 | } |
| 63 | for _, name := range names { |
| 64 | if strings.HasPrefix(name, "reboot-"+r.Authorization.ID+"-") && strings.HasSuffix(name, ".json") { |
| 65 | if err := ctx.Err(); err != nil { |
| 66 | return err |
| 67 | } |
| 68 | if err := l.state.Remove(name); err != nil && !errors.Is(err, os.ErrNotExist) { |
| 69 | return err |
| 70 | } |
| 71 | } |
| 72 | } |
| 73 | // Every record-lock user first takes namespace.lock. No waiter can hold |
| 74 | // this lock's old inode while another process creates its replacement. |
| 75 | // Remove it first: a lost reply leaves the retirement discoverable for a |
| 76 | // later pass, rather than leaving an orphan lock after record collection. |
| 77 | if err := l.state.Remove("retirement-" + r.Authorization.ID + ".lock"); err != nil && !errors.Is(err, os.ErrNotExist) { |
| 78 | return err |
| 79 | } |
| 80 | if err := ctx.Err(); err != nil { |
| 81 | return err |
| 82 | } |
| 83 | if err := l.state.Remove("retirement-" + r.Authorization.ID + ".json"); err != nil { |
| 84 | return err |
| 85 | } |
| 86 | return l.removeCollectedBinding(ctx, r.Authorization.ID) |
| 87 | } |
| 88 | |
| 89 | // Called under namespace.lock. A binding cannot exist before its retirement, |
| 90 | // and no retirement can be recreated while this lock is held. This removes |
| 91 | // only private metadata, never an artifact or its namespace binding. |
| 92 | func (l *servingParkLedger) removeCollectedBinding(ctx context.Context, id string) error { |
| 93 | if err := ctx.Err(); err != nil { |
| 94 | return err |
| 95 | } |
| 96 | if _, err := l.state.Read("retirement-" + id + ".json"); !errors.Is(err, os.ErrNotExist) { |
| 97 | return err |
| 98 | } |
| 99 | err := l.state.Remove("current-retirement-" + id + ".json") |
| 100 | if errors.Is(err, os.ErrNotExist) { |
| 101 | return nil |
| 102 | } |
| 103 | return err |
| 104 | } |
| 105 | |
| 106 | func (l *servingParkLedger) collectOrphanBindings(ctx context.Context) error { |
| 107 | names, err := l.state.Names() |
| 108 | if err != nil { |
| 109 | return err |
| 110 | } |
| 111 | for _, name := range names { |
| 112 | if !strings.HasPrefix(name, "current-retirement-") || !strings.HasSuffix(name, ".json") { |
| 113 | continue |
| 114 | } |
| 115 | id := strings.TrimSuffix(strings.TrimPrefix(name, "current-retirement-"), ".json") |
| 116 | if !stopDigest(id) { |
| 117 | return hostbox.ErrChanged |
| 118 | } |
| 119 | if err := l.removeCollectedBinding(ctx, id); err != nil { |
| 120 | return err |
| 121 | } |
| 122 | } |
| 123 | return nil |
| 124 | } |
| 125 | |
amadan