amadan

carlos / platform Public

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

Plain git — no account needed to clone.

Download

Download this file

1package maintenanceguard_test
2
3import (
4 "bytes"
5 "context"
6 "crypto/sha256"
7 "encoding/base64"
8 "encoding/hex"
9 "errors"
10 "fmt"
11 "os"
12 "sync"
13 "testing"
14 "time"
15
16 guard "github.com/carlosframework/platform/internal/maintenanceguard"
17 "github.com/carlosframework/platform/internal/release"
18)
19
20var ctx = context.Background()
21
22func id(b byte) string { return base64.RawURLEncoding.EncodeToString(bytes.Repeat([]byte{b}, 32)) }
23func hash(s string) string { x := sha256.Sum256([]byte(s)); return hex.EncodeToString(x[:]) }
24
25type fixture struct {
26 store *release.DirStore
27 repo *guard.Repository
28 identity guard.Identity
29}
30
31func setup(t *testing.T) fixture {
32 t.Helper()
33 s := release.NewDirStore(t.TempDir())
34 identity, e := guard.RecordIdentity("aaa", "home", "one.home.example", time.Date(2026, 9, 8, 0, 0, 0, 1, time.UTC))
35 if e != nil {
36 t.Fatal(e)
37 }
38 return fixture{s, guard.New(s), guard.Identity{Account: "aaa", App: "home", Host: "one.home.example", Record: identity}}
39}
40func (f fixture) ref(t *testing.T, kind guard.ReferenceKind, n byte) guard.Reference {
41 t.Helper()
42 r, e := f.repo.PutReference(ctx, f.identity.Host, kind, id(n), []byte(fmt.Sprintf(`{"metadata":"%s-%d"}`, kind, n)))
43 if e != nil {
44 t.Fatal(e)
45 }
46 return r
47}
48func (f fixture) claim(t *testing.T, n byte) guard.ClaimSnapshot {
49 t.Helper()
50 s, e := f.repo.AcquireClaim(ctx, "aaa", "home", id(n), guard.MaintenanceOperation)
51 if e != nil {
52 t.Fatal(e)
53 }
54 return s
55}
56func (f fixture) initial(t *testing.T, c guard.ClaimSnapshot) guard.Guard {
57 t.Helper()
58 return guard.Guard{Version: 1, Identity: f.identity, Generation: 1, Phase: guard.Requested, OperationID: c.Value().OperationID, OwnerBox: "box-one", RecordRevision: `"original-record-etag"`, Request: f.ref(t, guard.RequestReference, 2), Attempt: f.ref(t, guard.AttemptReference, 3), ArtifactVersion: "v1", ArtifactDigest: hash("artifact"), ConfigDigest: hash("randomized-config-envelope"), RequiredContracts: []string{}}
59}
60func (f fixture) create(t *testing.T, c guard.ClaimSnapshot) guard.GuardSnapshot {
61 t.Helper()
62 g, e := f.repo.CreateGuard(ctx, c, f.initial(t, c))
63 if e != nil {
64 t.Fatal(e)
65 }
66 return g
67}
68func (f fixture) step(t *testing.T, c guard.ClaimSnapshot, s guard.GuardSnapshot, phase guard.Phase) guard.GuardSnapshot {
69 t.Helper()
70 g := s.Value()
71 g.Generation++
72 g.Phase = phase
73 if phase == guard.ResumeRequested {
74 g.Request = f.ref(t, guard.RequestReference, 4)
75 g.Attempt = f.ref(t, guard.AttemptReference, 5)
76 g.RequiredContracts = []string{"home.operator.v1"}
77 g.Evidence = nil
78 }
79 if phase == guard.Succeeded || phase == guard.Resumed {
80 ref := f.ref(t, guard.ReceiptReference, byte(g.Generation+10))
81 g.Evidence = &ref
82 }
83 next, e := f.repo.AdvanceGuard(ctx, c, s, g)
84 if e != nil {
85 t.Fatal(phase, e)
86 }
87 return next
88}
89func (f fixture) resumed(t *testing.T, c guard.ClaimSnapshot) guard.GuardSnapshot {
90 s := f.create(t, c)
91 for _, p := range []guard.Phase{guard.Quiescing, guard.Fenced, guard.Executing, guard.Succeeded, guard.ResumeRequested, guard.Resumed} {
92 s = f.step(t, c, s, p)
93 }
94 return s
95}
96func denied(t *testing.T, f fixture, p *guard.Projection, contracts []string) {
97 t.Helper()
98 d, e := f.repo.CheckServing(ctx, f.identity, p, contracts)
99 if d != guard.Denied || !errors.Is(e, guard.ErrDenied) {
100 t.Fatal(d, e)
101 }
102}
103
104func TestGuardUnmanagedAndAbsenceRecheck(t *testing.T) {
105 f := setup(t)
106 if d, e := f.repo.CheckServing(ctx, f.identity, nil, nil); e != nil || d != guard.Unmanaged {
107 t.Fatal(d, e)
108 }
109 p := guard.Projection{Version: 1, Generation: 1, GuardDigest: hash("guard")}
110 denied(t, f, &p, nil)
111 c := f.claim(t, 1)
112 // A claim alone is not a host fence. This models the crash before guard creation.
113 if d, e := f.repo.CheckServing(ctx, f.identity, nil, nil); e != nil || d != guard.Unmanaged {
114 t.Fatal(d, e)
115 }
116 s := f.create(t, c)
117 p, e := s.Projection()
118 if e != nil {
119 t.Fatal(e)
120 }
121 // A worker that observed absence before acquiring its lease must recheck.
122 denied(t, f, nil, nil)
123 denied(t, f, &p, nil)
124}
125func TestGuardResumeCrashProjectionAndContracts(t *testing.T) {
126 f := setup(t)
127 c := f.claim(t, 1)
128 s := f.create(t, c)
129 for _, phase := range []guard.Phase{guard.Quiescing, guard.Fenced, guard.Executing, guard.Succeeded, guard.ResumeRequested} {
130 s = f.step(t, c, s, phase)
131 p, _ := s.Projection()
132 denied(t, f, &p, []string{"home.operator.v1"})
133 }
134 oldProjection, _ := s.Projection()
135 s = f.step(t, c, s, guard.Resumed)
136 // Durable guard publication without the record CAS must not open admission.
137 denied(t, f, &oldProjection, []string{"home.operator.v1"})
138 denied(t, f, nil, []string{"home.operator.v1"})
139 reopened := guard.New(f.store)
140 loaded, e := reopened.ReadGuard(ctx, f.identity.Host)
141 if e != nil {
142 t.Fatal(e)
143 }
144 p, _ := loaded.Projection()
145 if d, e := reopened.CheckServing(ctx, f.identity, &p, []string{"home.operator.v1"}); e != nil || d != guard.Managed {
146 t.Fatal(d, e)
147 }
148 denied(t, f, &p, nil)
149 denied(t, f, &p, []string{"home.operator.v0"})
150 foreign := f.identity
151 foreign.Account = "bbb"
152 if d, e := f.repo.CheckServing(ctx, foreign, &p, []string{"home.operator.v1"}); d != guard.Denied || e == nil {
153 t.Fatal("hostname reassigned")
154 }
155 foreign = f.identity
156 foreign.Record = hash("replacement")
157 if d, e := f.repo.CheckServing(ctx, foreign, &p, []string{"home.operator.v1"}); d != guard.Denied || e == nil {
158 t.Fatal("record replaced")
159 }
160 if _, e = f.repo.ReleaseClaim(ctx, c); e != nil {
161 t.Fatal(e)
162 }
163 held, e := f.repo.ReadLifecycle(ctx, "aaa", "home")
164 if e != nil || held.Value().State != "idle" || held.Value().Generation != 2 {
165 t.Fatal(held.Value(), e)
166 }
167}
168func TestGuardStaleWorkersAndExecutingRecovery(t *testing.T) {
169 f := setup(t)
170 c := f.claim(t, 1)
171 first := f.create(t, c)
172 second := f.step(t, c, first, guard.Quiescing)
173 stale := first.Value()
174 stale.Generation++
175 stale.Phase = guard.Quiescing
176 if _, e := f.repo.AdvanceGuard(ctx, c, first, stale); !errors.Is(e, release.ErrCASConflict) {
177 t.Fatal("stale write did not preserve CAS conflict", e)
178 }
179 s := f.step(t, c, second, guard.Fenced)
180 s = f.step(t, c, s, guard.Executing)
181 afterCrash, e := f.repo.ReadGuard(ctx, f.identity.Host)
182 if e != nil {
183 t.Fatal(e)
184 }
185 repeat := afterCrash.Value()
186 repeat.Generation++
187 if _, e = f.repo.AdvanceGuard(ctx, c, afterCrash, repeat); !errors.Is(e, guard.ErrConflict) {
188 t.Fatal("executing automatically repeated", e)
189 }
190 s = f.step(t, c, afterCrash, guard.Ambiguous)
191 retry := s.Value()
192 retry.Generation++
193 retry.Phase = guard.Requested
194 retry.Request = f.ref(t, guard.RequestReference, 8)
195 retry.Attempt = f.ref(t, guard.AttemptReference, 9)
196 retry.Evidence = nil
197 if _, e = f.repo.AdvanceGuard(ctx, c, s, retry); e != nil {
198 t.Fatal("explicit new attempt refused", e)
199 }
200}
201func TestGuardContinuityAndContractFloorCannotShrink(t *testing.T) {
202 f := setup(t)
203 c := f.claim(t, 1)
204 s := f.create(t, c)
205 for _, change := range []func(*guard.Guard){func(g *guard.Guard) { g.Generation-- }, func(g *guard.Guard) { g.Generation++ }, func(g *guard.Guard) { g.Identity.Record = hash("other") }, func(g *guard.Guard) { g.OwnerBox = "box-two" }, func(g *guard.Guard) { g.Request.Digest = hash("other") }, func(g *guard.Guard) { g.ArtifactDigest = hash("other") }} {
206 g := s.Value()
207 g.Generation++
208 g.Phase = guard.Quiescing
209 change(&g)
210 if _, e := f.repo.AdvanceGuard(ctx, c, s, g); !errors.Is(e, guard.ErrConflict) {
211 t.Fatal("continuity changed", e)
212 }
213 }
214 for _, p := range []guard.Phase{guard.Quiescing, guard.Fenced, guard.ResumeRequested, guard.Resumed} {
215 s = f.step(t, c, s, p)
216 }
217 if _, e := f.repo.ReleaseClaim(ctx, c); e != nil {
218 t.Fatal(e)
219 }
220 c = f.claim(t, 20)
221 g := s.Value()
222 g.Generation++
223 g.Phase = guard.Requested
224 g.OperationID = c.Value().OperationID
225 g.Request = f.ref(t, guard.RequestReference, 21)
226 g.Attempt = f.ref(t, guard.AttemptReference, 22)
227 g.Evidence = nil
228 g.RequiredContracts = nil
229 if _, e := f.repo.AdvanceGuard(ctx, c, s, g); !errors.Is(e, guard.ErrConflict) {
230 t.Fatal("floor removed", e)
231 }
232}
233func TestImmutableReferenceAndMissingEvidence(t *testing.T) {
234 f := setup(t)
235 r := f.ref(t, guard.RequestReference, 2)
236 if _, e := f.repo.PutReference(ctx, f.identity.Host, guard.RequestReference, r.ID, []byte("changed")); !errors.Is(e, guard.ErrConflict) {
237 t.Fatal(e)
238 }
239 c := f.claim(t, 1)
240 s := f.resumed(t, c)
241 p, _ := s.Projection()
242 // Private payload expiry is separate and cannot remove permanent metadata.
243 private := "private/expired-payload"
244 if _, e := f.store.Put(ctx, private, []byte("synthetic-private"), ""); e != nil {
245 t.Fatal(e)
246 }
247 if e := f.store.Delete(ctx, private); e != nil {
248 t.Fatal(e)
249 }
250 if d, e := f.repo.CheckServing(ctx, f.identity, &p, []string{"home.operator.v1"}); e != nil || d != guard.Managed {
251 t.Fatal(d, e)
252 }
253 key, _ := guard.ReceiptKey(f.identity.Host, s.Value().Evidence.ID)
254 if e := f.store.Delete(ctx, key); e != nil {
255 t.Fatal(e)
256 }
257 denied(t, f, &p, []string{"home.operator.v1"})
258}
259
260// absenceBarrier forces both contenders to observe absence before either CAS.
261type absenceBarrier struct {
262 release.Store
263 key string
264 mu sync.Mutex
265 n int
266 ready chan struct{}
267}
268
269func (s *absenceBarrier) Get(ctx context.Context, key string) ([]byte, string, error) {
270 b, etag, e := s.Store.Get(ctx, key)
271 if key == s.key && errors.Is(e, os.ErrNotExist) {
272 s.mu.Lock()
273 s.n++
274 if s.n == 2 {
275 close(s.ready)
276 }
277 s.mu.Unlock()
278 <-s.ready
279 }
280 return b, etag, e
281}
282func TestLifecycleSimultaneousAbsenceMaintenanceVersusPurge(t *testing.T) {
283 for i := 0; i < 20; i++ {
284 f := setup(t)
285 key, _ := guard.LifecycleKey("aaa", "home")
286 barrier := &absenceBarrier{Store: f.store, key: key, ready: make(chan struct{})}
287 r := guard.New(barrier)
288 record := "apps/aaa/home/instances/one.home.example.json"
289 if _, e := f.store.Put(ctx, record, []byte("existing-record"), ""); e != nil {
290 t.Fatal(e)
291 }
292 type result struct {
293 kind guard.OperationKind
294 claim guard.ClaimSnapshot
295 err error
296 }
297 out := make(chan result, 2)
298 for n, kind := range []guard.OperationKind{guard.MaintenanceOperation, guard.PurgeOperation} {
299 go func(n int, kind guard.OperationKind) {
300 c, e := r.AcquireClaim(ctx, "aaa", "home", id(byte(n+40)), kind)
301 out <- result{kind, c, e}
302 }(n, kind)
303 }
304 a, b := <-out, <-out
305 if (a.err == nil) == (b.err == nil) {
306 t.Fatal("two absence readers both won or lost", a.err, b.err)
307 }
308 winner, loser := a, b
309 if winner.err != nil {
310 winner, loser = b, a
311 }
312 if !errors.Is(loser.err, release.ErrCASConflict) {
313 t.Fatal(loser.err)
314 }
315 if winner.kind == guard.PurgeOperation {
316 if e := f.store.Delete(ctx, record); e != nil {
317 t.Fatal(e)
318 }
319 } else {
320 g := f.initial(t, winner.claim)
321 if _, e := r.CreateGuard(ctx, winner.claim, g); e != nil {
322 t.Fatal(e)
323 }
324 }
325 s, e := r.ReadGuard(ctx, f.identity.Host)
326 if e != nil {
327 t.Fatal(e)
328 }
329 _, _, recordErr := f.store.Get(ctx, record)
330 if winner.kind == guard.MaintenanceOperation && (!s.Exists() || recordErr != nil) {
331 t.Fatal("purge crossed maintenance claim")
332 }
333 if winner.kind == guard.PurgeOperation && (s.Exists() || !errors.Is(recordErr, os.ErrNotExist)) {
334 t.Fatal("maintenance crossed purge claim")
335 }
336 }
337}
338