amadan

carlos / platform Public

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

Plain git — no account needed to clone.

Download

Download this file

1package activator
2
3import (
4 "context"
5 "errors"
6 "fmt"
7 "os"
8 "os/exec"
9 "testing"
10 "time"
11
12 "github.com/carlosframework/platform/internal/hostbox"
13 "github.com/carlosframework/platform/internal/hostclaim"
14 "github.com/carlosframework/platform/internal/instancectl"
15 "github.com/carlosframework/platform/internal/registry"
16 "github.com/carlosframework/platform/internal/release"
17)
18
19func servingRetryFixture(t *testing.T) (release.Store, hostclaim.Observed, *ClaimRuntime, ServingRetries, *time.Time, *int) {
20 t.Helper()
21 ctx := context.Background()
22 st := release.NewDirStore(t.TempDir())
23 now := time.Date(2026, 10, 6, 12, 0, 0, 0, time.UTC)
24 host := "shop.bab.oncarlos.com"
25 in := hostclaim.NewHolder{Family: "instance", Key: "apps/bab/jam/instances/" + host + ".json", Account: "bab", App: "jam", Host: host, Region: "ie", RecordRegion: "ie", CreatedAt: now}
26 body := []byte(`{"schema":1,"host":"shop.bab.oncarlos.com","account":"bab","app":"jam","region":"ie","created_at":"2026-10-06T12:00:00Z"}`)
27 absent, _ := hostclaim.Read(ctx, st, host)
28 o, err := hostclaim.Update(ctx, st, absent, hostclaim.BeginCreate(in, body, "create", "console", now))
29 if err != nil {
30 t.Fatal(err)
31 }
32 o, err = hostclaim.Update(ctx, st, o, hostclaim.CompleteCreate("create", "box"))
33 if err != nil {
34 t.Fatal(err)
35 }
36 seq, stops := 0, 0
37 d := ServingRetries{Claims: st, Box: "box", Enrollment: "enrollment", BootID: "boot", ControllerID: "controller", Now: func() time.Time { return now }, NewID: func() string { seq++; return fmt.Sprint("id", seq) }, Inputs: func(context.Context, registry.Route) (string, error) { return "inputs", nil }, StopAttempt: func(context.Context, hostclaim.Observed, hostclaim.WakeRetryAttempt) error { stops++; return nil }, ExactHealth: func(context.Context, *exec.Cmd, hostclaim.WakeRetryAttempt) error { return nil }}
38 base := &ClaimRuntime{Box: "box", Read: func(ctx context.Context, host string) (hostclaim.Observed, error) {
39 return hostclaim.Read(ctx, st, host)
40 }, StopForeign: func(context.Context, string) error { return nil }, Prerequisites: func(context.Context, registry.Route, hostclaim.Observed, StartPolicy) error { return nil }, Admit: func(ctx context.Context, _ registry.Route, o hostclaim.Observed, _ StartPolicy) (hostclaim.Observed, error) {
41 r := retryOf(o)
42 if r == nil || r.Attempt == nil || r.Attempt.Session != 0 {
43 t.Fatal("admission preceded first reservation")
44 }
45 seq++
46 return hostclaim.Update(ctx, st, o, hostclaim.AdmitWriter(hostclaim.Admission{ObservedSession: o.Claim.Holder.Session, ObservedGeneration: o.Claim.Holder.Generation, Region: "ie", PredecessorRegion: "ie", Box: "box", Mode: "cold", NewGeneration: fmt.Sprint("gen", seq), PredecessorCertified: true}))
47 }, Prepare: func(context.Context, registry.Route, hostclaim.Observed, StartPolicy) (ClaimStart, error) {
48 return ClaimStart{}, nil
49 }, FailedAdmission: func(context.Context, hostclaim.Observed, error) error { return nil }}
50 return st, o, base, d, &now, &stops
51}
52func TestServingRetriesSurviveRestartAndFifthFailure(t *testing.T) {
53 st, o, base, d, now, stops := servingRetryFixture(t)
54 ctx := context.Background()
55 p := StartPolicy{}
56 rt := registry.Route{Host: o.Host}
57 for n := 1; n <= 5; n++ {
58 runtime, err := d.Wrap(base)
59 if err != nil {
60 t.Fatal(err)
61 }
62 if err = runtime.Prerequisites(ctx, rt, o, p); err != nil {
63 t.Fatal(err)
64 }
65 admitted, err := runtime.Admit(ctx, rt, o, p)
66 if err != nil {
67 t.Fatal(err)
68 }
69 if retryOf(admitted).Attempt.Session != admitted.Claim.Holder.Session {
70 t.Fatal("unbound start")
71 }
72 s, err := runtime.Prepare(ctx, rt, admitted, p)
73 if err != nil {
74 t.Fatal(err)
75 }
76 if err = s.Reserve(ctx, 0); err != nil {
77 t.Fatal(err)
78 }
79 if err = s.Failed(ctx, 0); err != nil {
80 t.Fatal(err)
81 }
82 o, err = hostclaim.Read(ctx, st, o.Host)
83 if err != nil {
84 t.Fatal(err)
85 }
86 if retryOf(o).AttemptsReserved != n {
87 t.Fatal("budget reset")
88 }
89 if err = runtime.Prerequisites(ctx, rt, o, p); err == nil {
90 t.Fatal("early retry admitted")
91 }
92 *now = now.Add(time.Second << uint(n-1))
93 }
94 if retryOf(o).State != "failed" || *stops != 5 {
95 t.Fatalf("retry=%+v stops=%d", retryOf(o), *stops)
96 }
97 runtime, _ := d.Wrap(base)
98 if err := runtime.Prerequisites(ctx, rt, o, p); err == nil {
99 t.Fatal("new controller reset failed latch")
100 }
101}
102func TestServingRetriesReadOnlyRefusalSpendsNothing(t *testing.T) {
103 st, o, base, d, _, _ := servingRetryFixture(t)
104 ctx := context.Background()
105 base.Prerequisites = func(context.Context, registry.Route, hostclaim.Observed, StartPolicy) error {
106 return hostbox.ErrUnsupported
107 }
108 runtime, _ := d.Wrap(base)
109 if err := runtime.Prerequisites(ctx, registry.Route{Host: o.Host}, o, StartPolicy{}); err == nil {
110 t.Fatal("missing prerequisite passed")
111 }
112 fresh, _ := hostclaim.Read(ctx, st, o.Host)
113 if fresh.ETag != o.ETag || retryOf(fresh) != nil {
114 t.Fatal("prerequisite spent authority")
115 }
116}
117func TestServingRetriesCrashMustStopBeforeNextReservation(t *testing.T) {
118 st, o, base, d, _, stops := servingRetryFixture(t)
119 ctx := context.Background()
120 rt := registry.Route{Host: o.Host}
121 runtime, _ := d.Wrap(base)
122 admitted, err := runtime.Admit(ctx, rt, o, StartPolicy{})
123 if err != nil {
124 t.Fatal(err)
125 }
126 restarted, _ := d.Wrap(base)
127 if err = restarted.Prerequisites(ctx, rt, admitted, StartPolicy{}); !errors.Is(err, hostbox.ErrChanged) {
128 t.Fatal(err)
129 }
130 fresh, _ := hostclaim.Read(ctx, st, o.Host)
131 if *stops != 1 || retryOf(fresh).AttemptsReserved != 1 || retryOf(fresh).State != "open" {
132 t.Fatalf("not settled: %+v", retryOf(fresh))
133 }
134}
135func TestServingRetriesHealthBindsExactLaunch(t *testing.T) {
136 st, o, base, d, _, _ := servingRetryFixture(t)
137 ctx := context.Background()
138 rt := registry.Route{Host: o.Host}
139 d.ExactHealth = func(context.Context, *exec.Cmd, hostclaim.WakeRetryAttempt) error { return hostbox.ErrChanged }
140 runtime, _ := d.Wrap(base)
141 admitted, err := runtime.Admit(ctx, rt, o, StartPolicy{})
142 if err != nil {
143 t.Fatal(err)
144 }
145 s, err := runtime.Prepare(ctx, rt, admitted, StartPolicy{})
146 if err != nil {
147 t.Fatal(err)
148 }
149 process, _ := os.FindProcess(os.Getpid())
150 if err = s.Healthy(ctx, 0, &exec.Cmd{Process: process}); err == nil {
151 t.Fatal("wrong invocation cleared episode")
152 }
153 fresh, _ := hostclaim.Read(ctx, st, o.Host)
154 if retryOf(fresh) == nil {
155 t.Fatal("health cleared episode")
156 }
157 d.ExactHealth = func(context.Context, *exec.Cmd, hostclaim.WakeRetryAttempt) error { return nil }
158 runtime, _ = d.Wrap(base)
159 s, err = runtime.Prepare(ctx, rt, fresh, StartPolicy{})
160 if err != nil {
161 t.Fatal(err)
162 }
163 if err = s.Healthy(ctx, 0, &exec.Cmd{Process: process}); err != nil {
164 t.Fatal(err)
165 }
166 fresh, _ = hostclaim.Read(ctx, st, o.Host)
167 if retryOf(fresh) != nil {
168 t.Fatal("exact health did not clear")
169 }
170}
171
172func TestServingDiskRefusalSpendsNoRetry(t *testing.T) {
173 st, o, base, d, _, stops := servingRetryFixture(t)
174 effects := 0
175 base.Prerequisites = func(context.Context, registry.Route, hostclaim.Observed, StartPolicy) error { effects++; return nil }
176 d.BeforeReserve = func(context.Context, registry.Route, hostclaim.Observed) error { return instancectl.ErrDiskFull }
177 runtime, err := d.Wrap(base)
178 if err != nil {
179 t.Fatal(err)
180 }
181 for i := 0; i < 12; i++ {
182 if err := runtime.Prerequisites(t.Context(), registry.Route{Host: o.Host}, o, StartPolicy{}); !errors.Is(err, instancectl.ErrDiskFull) {
183 t.Fatal(err)
184 }
185 if _, err := runtime.Admit(t.Context(), registry.Route{Host: o.Host}, o, StartPolicy{}); !errors.Is(err, instancectl.ErrDiskFull) {
186 t.Fatal(err)
187 }
188 }
189 after, err := hostclaim.Read(t.Context(), st, o.Host)
190 if err != nil || after.ETag != o.ETag || effects != 0 || *stops != 0 {
191 t.Fatal("low disk spent a retry or stopped a writer", err, effects, *stops)
192 }
193}
194
195func TestServingSeedRerunSettlesWithoutReservation(t *testing.T) {
196 for _, fault := range []string{"", "stop-fails", "stale"} {
197 t.Run(fault, func(t *testing.T) {
198 st, o, base, d, _, stops := servingRetryFixture(t)
199 runtime, err := d.Wrap(base)
200 if err != nil {
201 t.Fatal(err)
202 }
203 admitted, err := runtime.Admit(t.Context(), registry.Route{Host: o.Host}, o, StartPolicy{})
204 if err != nil {
205 t.Fatal(err)
206 }
207 expected := admitted.Claim.Holder.WakeRetry.AttemptsReserved
208 if fault == "stop-fails" {
209 d.StopAttempt = func(context.Context, hostclaim.Observed, hostclaim.WakeRetryAttempt) error { return hostbox.ErrChanged }
210 }
211 if fault == "stale" {
212 admitted.ETag = "stale"
213 }
214 d.BeforeReserve = func(context.Context, registry.Route, hostclaim.Observed) error {
215 t.Fatal("settlement required new preparation capacity")
216 return nil
217 }
218 settled, err := d.SettleAbandoned(t.Context(), admitted)
219 if fault != "" {
220 if err == nil {
221 t.Fatal("settled", fault)
222 }
223 return
224 }
225 r := settled.Claim.Holder.WakeRetry
226 if err != nil || r.State != "open" || r.Attempt != nil || r.AttemptsReserved != expected || *stops != 1 {
227 t.Fatal("rerun settlement", r, err, *stops)
228 }
229 again, err := d.SettleAbandoned(t.Context(), settled)
230 if err != nil || again.ETag != settled.ETag || *stops != 1 {
231 t.Fatal("settled replay changed state", err)
232 }
233 fresh, err := hostclaim.Read(t.Context(), st, o.Host)
234 if err != nil || fresh.ETag != settled.ETag {
235 t.Fatal(err)
236 }
237 })
238 }
239}
240
241func TestServingRetriesReturnOwnClaimWrites(t *testing.T) {
242 st, o, base, d, now, _ := servingRetryFixture(t)
243 var expected hostclaim.Observed
244 calls := 0
245 base.Prepare = func(_ context.Context, _ registry.Route, o hostclaim.Observed, _ StartPolicy) (ClaimStart, error) {
246 expected = o
247 return ClaimStart{RetryClaimWrite: func(before, after hostclaim.Observed) error {
248 if before.ETag != expected.ETag || after.ETag == before.ETag || after.Claim.Gen != before.Claim.Gen+1 {
249 t.Fatal("write did not chain from Prepare")
250 }
251 expected = after
252 calls++
253 return nil
254 }}, nil
255 }
256 runtime, err := d.Wrap(base)
257 if err != nil {
258 t.Fatal(err)
259 }
260 rt := registry.Route{Host: o.Host}
261 admitted, err := runtime.Admit(t.Context(), rt, o, StartPolicy{})
262 if err != nil {
263 t.Fatal(err)
264 }
265 start, err := runtime.Prepare(t.Context(), rt, admitted, StartPolicy{})
266 if err != nil {
267 t.Fatal(err)
268 }
269 if err := start.Reserve(t.Context(), 0); err != nil {
270 t.Fatal(err)
271 }
272 if calls != 0 {
273 t.Fatal("first reservation invented a write")
274 }
275 if err := start.Failed(t.Context(), 0); err != nil {
276 t.Fatal(err)
277 }
278 *now = now.Add(2 * time.Second)
279 if err := start.Reserve(t.Context(), 1); err != nil {
280 t.Fatal(err)
281 }
282 actual, err := hostclaim.Read(t.Context(), st, o.Host)
283 if err != nil || calls != 2 || expected.ETag != actual.ETag {
284 t.Fatal("lost retry write", calls, err)
285 }
286}
287