amadan

carlos / platform Public

Clone 
git clone https://amadan.net/carlos/platform

Plain git — no account needed to clone.

Download

Download this file

1package hostops
2
3import (
4 "context"
5 "encoding/json"
6 "errors"
7 "os"
8 "sort"
9 "strings"
10
11 "github.com/carlosframework/platform/internal/hostclaim"
12 "github.com/carlosframework/platform/internal/lease"
13 "github.com/carlosframework/platform/internal/release"
14)
15
16type PurgeObject struct {
17 Key string `json:"key"`
18 ETag string `json:"etag"`
19 VersionID string `json:"version_id,omitempty"`
20 Body json.RawMessage `json:"body"`
21}
22type PurgeHost struct {
23 ClaimETag string `json:"claim_etag,omitempty"`
24 ClaimGen uint64 `json:"claim_gen,omitempty"`
25 Holder *hostclaim.Holder `json:"holder,omitempty"`
26 Registrations map[string]json.RawMessage `json:"registrations,omitempty"`
27 RecordETag string `json:"record_etag,omitempty"`
28 RecordVersion string `json:"record_version,omitempty"`
29 FenceState string `json:"fence_state,omitempty"`
30 FenceGen uint64 `json:"fence_gen,omitempty"`
31 FenceETag string `json:"fence_etag,omitempty"`
32 Retired bool `json:"retired"`
33 FreeFromGen uint64 `json:"free_from_gen,omitempty"`
34}
35type PurgeSweep struct {
36 Batch uint64 `json:"batch"`
37 Bucket string `json:"bucket"`
38 Versions []lease.ObjectVersion `json:"versions"`
39 Prepared bool `json:"prepared"`
40 Deleted bool `json:"deleted"`
41 Verified bool `json:"verified"`
42}
43type PurgeCleanup struct {
44 Sweep PurgeSweep `json:"sweep"`
45 Tombstone bool `json:"tombstone"`
46 AppRow bool `json:"app_row"`
47 Fences bool `json:"fences"`
48 Admission bool `json:"admission"`
49}
50type PurgeJournal struct {
51 IngestRegistrations map[string]Registration `json:"ingest_registrations,omitempty"`
52 Purge
53 V int `json:"v"`
54 AppID string `json:"app_id"`
55 Account string `json:"account"`
56 App string `json:"app"`
57 RequestedBy string `json:"requested_by"`
58 RequestedAt string `json:"requested_at"`
59 Reason string `json:"reason"`
60 Tombstone PurgeObject `json:"tombstone"`
61 AdmissionEpoch uint64 `json:"admission_epoch"`
62 Hosts map[string]PurgeHost `json:"hosts"`
63 Effects map[string]json.RawMessage `json:"effects"`
64 Cleanup PurgeCleanup `json:"cleanup"`
65}
66
67func ReadPurgeJournal(ctx context.Context, s release.Store, account, app, id, appID string) (PurgeJournal, string, error) {
68 key, err := PurgeKey(account, app)
69 if err != nil {
70 return PurgeJournal{}, "", err
71 }
72 body, tag, err := s.Get(ctx, key)
73 if err != nil {
74 return PurgeJournal{}, "", err
75 }
76 var j PurgeJournal
77 if err = decode(body, &j); err != nil {
78 return j, "", err
79 }
80 if tag == "" || j.V != 1 || j.ID != id || j.AppID != appID || j.Account != account || j.App != app || phaseIndex(j.Phase) < 0 || j.Hosts == nil {
81 return j, "", ErrIdentity
82 }
83 if j.Tombstone.Key != release.AppTombstoneKey(account, app) || j.Tombstone.ETag == "" || !json.Valid(j.Tombstone.Body) {
84 return j, "", ErrIdentity
85 }
86 for id, reg := range j.IngestRegistrations {
87 if !validIngestRegistration(id, reg, j.AppID, j.AdmissionEpoch) {
88 return j, "", ErrIdentity
89 }
90 }
91 return j, tag, nil
92}
93func purgeExcluded(j PurgeJournal, key string) bool {
94 prefix := "apps/" + j.Account + "/" + j.App + "/"
95 return key == prefix+"admission.json" || key == prefix+"purge.json" || key == prefix+"purge-request.json" || key == j.Tombstone.Key
96}
97func PreparePurgeSweep(ctx context.Context, s release.Store, w *ControlWriter, v lease.VersionedStore, bucket, account, app, id, appID string) error {
98 if v == nil || bucket == "" {
99 return ErrContract
100 }
101 j, tag, err := ReadPurgeJournal(ctx, s, account, app, id, appID)
102 if err != nil {
103 return err
104 }
105 if j.Phase != "retired" {
106 return ErrPending
107 }
108 if j.Cleanup.Sweep.Prepared {
109 if j.Cleanup.Sweep.Bucket != bucket {
110 return ErrIdentity
111 }
112 if !j.Cleanup.Sweep.Deleted || j.Cleanup.Sweep.Verified {
113 return nil
114 }
115 // A late old-app auxiliary write needs another persisted exact-version
116 // batch. Only the still-retired matching reservation can save it; a
117 // stale worker cannot enumerate/delete under done or a recreated app.
118 }
119 prefix := "apps/" + account + "/" + app + "/"
120 versions, err := v.ListVersions(ctx, bucket, prefix)
121 if err != nil {
122 return err
123 }
124 if j.Cleanup.Sweep.Batch == ^uint64(0) {
125 return ErrIdentity
126 }
127 batch := PurgeSweep{Batch: j.Cleanup.Sweep.Batch + 1, Bucket: bucket, Prepared: true, Versions: []lease.ObjectVersion{}}
128 for _, o := range versions {
129 if !strings.HasPrefix(o.Key, prefix) || o.VersionID == "" || o.VersionID == "null" {
130 return ErrIdentity
131 }
132 if purgeExcluded(j, o.Key) {
133 continue
134 }
135 tail := strings.TrimPrefix(o.Key, prefix)
136 if strings.HasPrefix(tail, "instances/") || strings.HasPrefix(tail, "sidecars/") {
137 // Historical record versions may be swept only after current records
138 // have been retired. The generic sweep never removes a live record.
139 if _, _, e := s.Get(ctx, o.Key); !errors.Is(e, os.ErrNotExist) {
140 if e != nil {
141 return e
142 }
143 return ErrPending
144 }
145 }
146 batch.Versions = append(batch.Versions, o)
147 }
148 j.Cleanup.Sweep = batch
149 _, err = w.SavePurgeJournal(ctx, j, tag)
150 return err
151}
152
153type retiredRecordSweepKey struct{}
154
155func DeletePurgeSweep(ctx context.Context, s release.Store, w *ControlWriter, v lease.VersionedStore, bucket, account, app, id, appID string) error {
156 if v == nil {
157 return ErrContract
158 }
159 j, _, err := ReadPurgeJournal(ctx, s, account, app, id, appID)
160 if err != nil {
161 return err
162 }
163 if j.Phase != "retired" || !j.Cleanup.Sweep.Prepared || j.Cleanup.Sweep.Bucket != bucket {
164 return ErrPending
165 }
166 if j.Cleanup.Sweep.Deleted {
167 return nil
168 }
169 batch := append([]lease.ObjectVersion(nil), j.Cleanup.Sweep.Versions...)
170 // Replay old journals safely too: retire record data before markers.
171 sort.SliceStable(batch, func(i, k int) bool {
172 a := provisioningKey(batch[i].Key) && batch[i].DeleteMarker
173 b := provisioningKey(batch[k].Key) && batch[k].DeleteMarker
174 return !a && b
175 })
176 for _, o := range batch {
177 fresh, _, err := ReadPurgeJournal(ctx, s, account, app, id, appID)
178 if err != nil {
179 return err
180 }
181 if fresh.Phase != "retired" || fresh.Cleanup.Sweep.Bucket != bucket || fresh.Cleanup.Sweep.Batch != j.Cleanup.Sweep.Batch {
182 return ErrPending
183 }
184 if o.VersionID == "" || o.VersionID == "null" || purgeExcluded(j, o.Key) {
185 return ErrIdentity
186 }
187 deleteCtx := ctx
188 if provisioningKey(o.Key) {
189 // Recheck after preparation, before every version deletion. The
190 // existing exclusive purge authority excludes recreation.
191 if err := retiredRecordVersion(ctx, s, v, bucket, o.Key, o.VersionID); err != nil {
192 return err
193 }
194 deleteCtx = context.WithValue(ctx, retiredRecordSweepKey{}, o.Key)
195 }
196 if err = v.DeleteVersion(deleteCtx, bucket, o.Key, o.VersionID); err != nil {
197 return err
198 }
199 }
200 fresh, tag, err := ReadPurgeJournal(ctx, s, account, app, id, appID)
201 if err != nil {
202 return err
203 }
204 if fresh.Phase != "retired" || fresh.Cleanup.Sweep.Batch != j.Cleanup.Sweep.Batch {
205 return ErrPending
206 }
207 fresh.Cleanup.Sweep.Deleted = true
208 _, err = w.SavePurgeJournal(ctx, fresh, tag)
209 return err
210}
211func VerifyPurgeSweep(ctx context.Context, s release.Store, w *ControlWriter, v lease.VersionedStore, bucket, account, app, id, appID string) error {
212 j, tag, err := ReadPurgeJournal(ctx, s, account, app, id, appID)
213 if err != nil {
214 return err
215 }
216 if j.Phase != "retired" || !j.Cleanup.Sweep.Deleted || j.Cleanup.Sweep.Bucket != bucket {
217 return ErrPending
218 }
219 if j.Cleanup.Sweep.Verified {
220 return nil
221 }
222 prefix := "apps/" + account + "/" + app + "/"
223 versions, err := v.ListVersions(ctx, bucket, prefix)
224 if err != nil {
225 return err
226 }
227 for _, o := range versions {
228 if !strings.HasPrefix(o.Key, prefix) || !purgeExcluded(j, o.Key) {
229 return ErrPending
230 }
231 }
232 keys, err := s.List(ctx, prefix)
233 if err != nil {
234 return err
235 }
236 for _, key := range keys {
237 if !purgeExcluded(j, key) {
238 return ErrPending
239 }
240 }
241 j.Cleanup.Sweep.Verified = true
242 _, err = w.SavePurgeJournal(ctx, j, tag)
243 return err
244}
245