amadan

carlos / platform Public

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

Plain git — no account needed to clone.

Download

Download this file

1package hostbox
2
3import (
4 "context"
5 "errors"
6 "fmt"
7 "testing"
8 "time"
9
10 "github.com/carlosframework/platform/internal/hostclaim"
11 "github.com/carlosframework/platform/internal/lease"
12 "github.com/carlosframework/platform/internal/lease/leasetest"
13 "github.com/carlosframework/platform/internal/park"
14 "github.com/carlosframework/platform/internal/release"
15)
16
17type enrollmentFunc func(context.Context, string, string, time.Duration) error
18
19func (f enrollmentFunc) VerifyRegional(ctx context.Context, region, bucket string, limit time.Duration) error {
20 return f(ctx, region, bucket, limit)
21}
22
23type evidenceFunc func(context.Context, hostclaim.Observed, []lease.ObjectVersion) (string, time.Time, error)
24
25func (f evidenceFunc) VerifyRetirement(ctx context.Context, o hostclaim.Observed, v []lease.ObjectVersion) (string, time.Time, error) {
26 return f(ctx, o, v)
27}
28
29type workerLease struct{}
30
31func (workerLease) Check(context.Context) error { return nil }
32func (workerLease) Release(context.Context) error { return nil }
33func (workerLease) Close() {}
34
35type workerLeases struct{}
36
37func (workerLeases) Acquire(context.Context, string, string) (park.HeldLease, error) {
38 return workerLease{}, nil
39}
40
41func TestRegionalWorkerRealEngineCensusSweepAndBound(t *testing.T) {
42 ctx := t.Context()
43 evidenceAt := time.Date(2026, 10, 6, 10, 0, 0, 0, time.UTC)
44 host := "app.example"
45 inc, err := hostclaim.Incarnation("acct", "app", host, evidenceAt)
46 if err != nil {
47 t.Fatal(err)
48 }
49 location := "instances/" + host + "/" + inc
50 bucket := leasetest.NewVersionedBucket()
51 bucket.AddBucket("data")
52 bucket.PageSize = 1
53 bucket.Put("data", location+"/0000/db", []byte("old"))
54 // The moves namespace participates in the top-level census independently.
55 bucket.Put("data", "moves/"+host+"/landing/lock.json", []byte("lease"))
56 claims := release.NewDirStore(t.TempDir())
57 _, err = hostclaim.Update(ctx, claims, hostclaim.Observed{Host: host, Absent: true}, func(c *hostclaim.Claim) error {
58 c.Holder = &hostclaim.Holder{Family: "instance", Key: "apps/acct/app/instances/" + host + ".json", Account: "acct", App: "app", Incarnation: inc, CreatedAt: evidenceAt, Location: location, Stamp: 1, Session: 1}
59 c.Op = hostclaim.Op{Kind: hostclaim.KindServing, Exec: "box"}
60 c.Touched = hostclaim.Touched{Complete: true, Regions: []string{"ie"}}
61 return nil
62 })
63 if err != nil {
64 t.Fatal(err)
65 }
66 seq := 0
67 var limits []time.Duration
68 engine := park.Engine{
69 Store: bucket, Claims: claims, Bucket: "data", Region: "ie", Leases: workerLeases{},
70 Now: func() time.Time { return evidenceAt.Add(18 * time.Minute) },
71 NewID: func() string { seq++; return fmt.Sprintf("id%d", seq) },
72 View: func(_ context.Context, o hostclaim.Observed) (park.View, error) {
73 return park.View{ClaimETag: o.ETag, Region: "ie", AuthoritativeRegion: "uk"}, nil
74 },
75 Authorize: func(_ context.Context, _ hostclaim.Observed, o hostclaim.Obligation) error {
76 if o.Park != nil {
77 limits = append(limits, o.Park.RequestLimit)
78 }
79 return nil
80 },
81 }
82 engine.VerifyMovesSweep = func(_ context.Context, o hostclaim.Observed, versions []lease.ObjectVersion) (park.MoveSweepProof, error) {
83 if len(versions) == 0 {
84 t.Fatal("missing census")
85 }
86 return park.MoveSweepProof{Host: o.Host, ClaimETag: o.ETag, MoveID: "retired", Incarnation: inc, RetiredAt: evidenceAt}, nil
87 }
88 worker := RegionalWorker{
89 Regional: Regional{Engine: engine, Owner: func(context.Context, hostclaim.Observed) error { return nil }},
90 Enrollment: enrollmentFunc(func(_ context.Context, region, gotBucket string, limit time.Duration) error {
91 if region != "ie" || gotBucket != "data" {
92 t.Fatalf("binding = %s/%s", region, gotBucket)
93 }
94 if limit != 120*time.Second {
95 t.Fatalf("enforced request limit = %v", limit)
96 }
97 return nil
98 }),
99 Evidence: evidenceFunc(func(_ context.Context, o hostclaim.Observed, versions []lease.ObjectVersion) (string, time.Time, error) {
100 if o.ETag == "" || len(versions) != 2 {
101 t.Fatalf("evidence observation = etag %q, versions %d", o.ETag, len(versions))
102 }
103 return inc, evidenceAt, nil
104 }),
105 }
106 if err := worker.Run(ctx); !errors.Is(err, park.ErrPending) {
107 t.Fatalf("new moves sweep must wait from listing time: %v", err)
108 }
109 if len(limits) == 0 {
110 t.Fatal("authorization did not see a scoped chore")
111 }
112 for _, limit := range limits {
113 if limit != RequestLimit {
114 t.Fatalf("request limit = %v", limit)
115 }
116 }
117}
118
119func TestRegionalWorkerRefusesMissingAuthorityEvidenceAndUnknownKinds(t *testing.T) {
120 base := RegionalWorker{}
121 if _, err := base.Census(t.Context()); !errors.Is(err, ErrUnsupported) {
122 t.Fatalf("missing providers: %v", err)
123 }
124
125 claims := release.NewDirStore(t.TempDir())
126 host := "unknown.example"
127 _, err := hostclaim.Update(t.Context(), claims, hostclaim.Observed{Host: host, Absent: true}, func(c *hostclaim.Claim) error {
128 c.Obligations = []hostclaim.Obligation{{ID: "id", Op: "op", Kind: "park-superseded", Region: "ie", Owner: "region:ie"}}
129 return nil
130 })
131 if err == nil {
132 // Claim validation itself currently refuses this pending-decision kind,
133 // which is the desired boundary. If it becomes schema-valid, runHost's
134 // explicit kind check below remains the adapter refusal.
135 w := RegionalWorker{Regional: Regional{Engine: park.Engine{Claims: claims, Region: "ie", Bucket: "data"}}, Enrollment: enrollmentFunc(func(context.Context, string, string, time.Duration) error { return nil })}
136 if got := w.runHost(t.Context(), host); !errors.Is(got, ErrUnsupported) {
137 t.Fatalf("unknown kind: %v", got)
138 }
139 }
140}
141
142func TestRegionalWorkerCensusIncludesQueuedClaimWithoutResidue(t *testing.T) {
143 for name, namespaces := range map[string][]string{
144 "direct-only": nil,
145 "evidence": {"evidence/fence/receipt.json"},
146 "exec": {"exec/controller.json"},
147 "both": {"evidence/fence/receipt.json", "exec/controller.json"},
148 } {
149 t.Run(name, func(t *testing.T) {
150 claims := release.NewDirStore(t.TempDir())
151 host := "queued.example"
152 created := time.Unix(1, 0).UTC()
153 inc, err := hostclaim.Incarnation("bab", "app", host, created)
154 if err != nil {
155 t.Fatal(err)
156 }
157 _, err = hostclaim.Update(t.Context(), claims, hostclaim.Observed{Host: host, Absent: true}, func(c *hostclaim.Claim) error {
158 c.Holder = &hostclaim.Holder{Family: "instance", Key: "apps/bab/app/instances/" + host + ".json", Account: "bab", App: "app", Incarnation: inc, CreatedAt: created, Location: "instances/" + host + "/" + inc, Stamp: 1}
159 c.Op = hostclaim.Op{Kind: hostclaim.KindParking, ID: "op"}
160 c.Obligations = []hostclaim.Obligation{{ID: "id", Op: "op", Kind: "park-replica", Region: "ie", Owner: "region:ie", Retirement: "retire", Park: &hostclaim.ParkScope{Location: c.Holder.Location, Incarnation: inc, EvidenceAt: time.Unix(1, 0).UTC(), RequestLimit: RequestLimit}}}
161 c.Touched = hostclaim.Touched{Complete: true, Regions: []string{"ie"}}
162 return nil
163 })
164 if err != nil {
165 t.Fatal(err)
166 }
167 for _, namespace := range namespaces {
168 for _, nestedHost := range []string{host, "nested-only.example"} {
169 key := "control/hosts/" + nestedHost + "/" + namespace
170 if _, err := claims.Put(t.Context(), key, []byte(`{"not_a_claim":true}`), ""); err != nil {
171 t.Fatal(err)
172 }
173 }
174 }
175 keys, err := claims.List(t.Context(), "control/hosts/")
176 if err != nil || len(keys) != 1+2*len(namespaces) {
177 t.Fatalf("recursive claim-store listing = %v, %v", keys, err)
178 }
179 bucket := leasetest.NewVersionedBucket()
180 bucket.AddBucket("data")
181 w := RegionalWorker{
182 Regional: Regional{Engine: park.Engine{Claims: claims, Store: bucket, Bucket: "data", Region: "ie"}},
183 Enrollment: enrollmentFunc(func(context.Context, string, string, time.Duration) error { return nil }),
184 Evidence: evidenceFunc(func(context.Context, hostclaim.Observed, []lease.ObjectVersion) (string, time.Time, error) {
185 return "", time.Time{}, errors.New("unused")
186 }),
187 }
188 hosts, err := w.Census(t.Context())
189 if err != nil || len(hosts) != 1 || hosts[0] != host {
190 t.Fatalf("census = %v, %v", hosts, err)
191 }
192 })
193 }
194}
195
196func TestRegionalWorkerLoopRunsStartupAndJoinsCancellation(t *testing.T) {
197 ctx, cancel := context.WithCancel(t.Context())
198 claims := release.NewDirStore(t.TempDir())
199 bucket := leasetest.NewVersionedBucket()
200 bucket.AddBucket("data")
201 started := make(chan struct{}, 1)
202 w := RegionalWorker{
203 Regional: Regional{Engine: park.Engine{Claims: claims, Store: bucket, Bucket: "data", Region: "ie"}},
204 Enrollment: enrollmentFunc(func(context.Context, string, string, time.Duration) error {
205 select {
206 case started <- struct{}{}:
207 default:
208 }
209 return nil
210 }),
211 Evidence: evidenceFunc(func(context.Context, hostclaim.Observed, []lease.ObjectVersion) (string, time.Time, error) {
212 return "", time.Time{}, errors.New("unused")
213 }),
214 }
215 done := make(chan error, 1)
216 go func() { done <- w.Loop(ctx, nil, nil) }()
217 <-started
218 cancel()
219 if err := <-done; !errors.Is(err, context.Canceled) {
220 t.Fatalf("loop = %v", err)
221 }
222}
223
224func TestRegionalWorkerDoesNotBindEvidenceToLateVersion(t *testing.T) {
225 host := "late.example"
226 inc := "aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa"
227 claims := release.NewDirStore(t.TempDir())
228 if _, err := hostclaim.Update(t.Context(), claims, hostclaim.Observed{Host: host, Absent: true}, func(*hostclaim.Claim) error { return nil }); err != nil {
229 t.Fatal(err)
230 }
231 bucket := leasetest.NewVersionedBucket()
232 bucket.AddBucket("data")
233 key := "instances/" + host + "/old/data"
234 bucket.Put("data", key, []byte("evidenced"))
235 seq := 0
236 engine := park.Engine{
237 Store: bucket, Claims: claims, Bucket: "data", Region: "ie", Leases: workerLeases{},
238 Now: func() time.Time { return time.Unix(2000, 0).UTC() },
239 NewID: func() string { seq++; return fmt.Sprintf("id%d", seq) },
240 View: func(_ context.Context, obs hostclaim.Observed) (park.View, error) {
241 return park.View{ClaimETag: obs.ETag, Region: "ie", AuthoritativeRegion: "uk"}, nil
242 },
243 Authorize: func(context.Context, hostclaim.Observed, hostclaim.Obligation) error { return nil },
244 }
245 w := RegionalWorker{
246 Regional: Regional{Engine: engine, Owner: func(context.Context, hostclaim.Observed) error { return nil }},
247 Enrollment: enrollmentFunc(func(context.Context, string, string, time.Duration) error { return nil }),
248 Evidence: evidenceFunc(func(context.Context, hostclaim.Observed, []lease.ObjectVersion) (string, time.Time, error) {
249 bucket.Put("data", key, []byte("late"))
250 return inc, time.Unix(1, 0).UTC(), nil
251 }),
252 }
253 if err := w.Run(t.Context()); !errors.Is(err, park.ErrChanged) {
254 t.Fatalf("Run = %v", err)
255 }
256 obs, err := hostclaim.Read(t.Context(), claims, host)
257 if err != nil || len(obs.Claim.Obligations) != 0 {
258 t.Fatalf("late residue queued under old evidence: %+v, %v", obs.Claim.Obligations, err)
259 }
260}
261
262func TestRegionalWorkerJoinsHolderlessSweepFromAnotherRegion(t *testing.T) {
263 ctx := t.Context()
264 host := "joining.example"
265 claims := release.NewDirStore(t.TempDir())
266 at := time.Unix(1, 0).UTC()
267 inc := "aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa"
268 uk := hostclaim.Obligation{ID: "uk-chore", Op: "original-op", Kind: "park-replica", Region: "uk", Owner: "region:uk", Retirement: "uk-retire", Park: &hostclaim.ParkScope{Location: "instances/" + host + "/" + inc, Incarnation: inc, EvidenceAt: at, RequestLimit: RequestLimit}}
269 if _, err := hostclaim.Update(ctx, claims, hostclaim.Observed{Host: host, Absent: true}, func(c *hostclaim.Claim) error {
270 c.Op = hostclaim.Op{Kind: hostclaim.KindParking, ID: "original-op", Intent: map[string]string{"park_sweep": "v1"}}
271 c.Obligations = []hostclaim.Obligation{uk}
272 return nil
273 }); err != nil {
274 t.Fatal(err)
275 }
276 bucket := leasetest.NewVersionedBucket()
277 bucket.AddBucket("data")
278 bucket.Put("data", "instances/"+host+"/"+inc+"/old", []byte("residue"))
279 seq := 0
280 seen := false
281 engine := park.Engine{Claims: claims, Store: bucket, Bucket: "data", Region: "ie", Leases: workerLeases{}, Now: func() time.Time { return at.Add(18 * time.Minute) }, NewID: func() string { seq++; return fmt.Sprintf("new%d", seq) }, View: func(_ context.Context, o hostclaim.Observed) (park.View, error) {
282 return park.View{ClaimETag: o.ETag, Region: "ie"}, nil
283 }, Authorize: func(_ context.Context, _ hostclaim.Observed, o hostclaim.Obligation) error {
284 if o.Op != "original-op" {
285 t.Fatalf("replaced operation: %+v", o)
286 }
287 seen = true
288 return nil
289 }}
290 w := RegionalWorker{Regional: Regional{Engine: engine, Owner: func(context.Context, hostclaim.Observed) error { return nil }}, Enrollment: enrollmentFunc(func(context.Context, string, string, time.Duration) error { return nil }), Evidence: evidenceFunc(func(context.Context, hostclaim.Observed, []lease.ObjectVersion) (string, time.Time, error) {
291 return inc, at, nil
292 })}
293 if err := w.Run(ctx); err != nil {
294 t.Fatal(err)
295 }
296 obs, err := hostclaim.Read(ctx, claims, host)
297 if err != nil || !seen || obs.Claim.Op.ID != "original-op" {
298 t.Fatalf("join: seen=%v obs=%+v err=%v", seen, obs, err)
299 }
300 found := false
301 for _, o := range obs.Claim.Obligations {
302 if o.ID == uk.ID {
303 found = true
304 }
305 }
306 if !found {
307 t.Fatal("other region's chore lost")
308 }
309}
310