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 "bytes"
5 "context"
6 "encoding/json"
7 "errors"
8 "fmt"
9 "strings"
10 "testing"
11
12 "github.com/carlosframework/platform/internal/provision"
13 "github.com/carlosframework/platform/internal/release"
14)
15
16func TestPlan9B8SequenceEveryRecordWriter(t *testing.T) {
17 ctx := context.Background()
18 f := newMoveStartFixture(t, nil)
19 records, err := NewConditionalRecordStore(f.raw, f.pub)
20 if err != nil {
21 t.Fatal(err)
22 }
23 key := f.raw.recordKey
24 // Ordinary typed writer: old absent sequence becomes one and then two.
25 for want := uint64(1); want <= 2; want++ {
26 if err := provision.SetRecordRegion(ctx, records, "bab", "app", f.host, "ie"); err != nil {
27 t.Fatal(err)
28 }
29 body, _, _ := records.Get(ctx, key)
30 rec, err := provision.DecodeRecord(body)
31 if err != nil || rec.RecordSeq != want {
32 t.Fatalf("ordinary: %+v %v", rec, err)
33 }
34 }
35 // Both effect branches also sequence their writes, including PutFrom.
36 for _, resource := range []string{key, "control-object:" + key} {
37 body, tag, _ := records.Get(ctx, key)
38 _, seq, _ := recordSequence(body)
39 permitted := context.WithValue(ctx, effectPermitKey{}, effectPermit{Resource: resource})
40 if _, err := release.PutObjectFrom(permitted, records, key, bytes.NewReader(body), int64(len(body)), tag); err != nil {
41 t.Fatal(err)
42 }
43 next, _, _ := records.Get(ctx, key)
44 _, got, _ := recordSequence(next)
45 if got != seq+1 {
46 t.Fatalf("effect %s: seq %d -> %d", resource, seq, got)
47 }
48 }
49 // Lifecycle updates keep the same counter after the first journal Put.
50 r := newResolveFixture(t, true)
51 if got, err := ResolveMoveStart(ctx, r.h, r.moving, moveStartID); err != nil || got != OutcomeComplete {
52 t.Fatalf("resolve: %q %v", got, err)
53 }
54 lifecycle, err := NewConditionalRecordStore(r.raw, r.pub)
55 if err != nil {
56 t.Fatal(err)
57 }
58 if rec, err := setLifecycleRecordMove(ctx, lifecycle, r.moving, func(rec *provision.Record) error { return nil }); err != nil || rec.RecordSeq != 2 {
59 t.Fatalf("returned lifecycle record: %+v %v", rec, err)
60 }
61 body, _ := r.record(t)
62 _, seq, err := recordSequence(body)
63 if err != nil || seq != 2 {
64 t.Fatalf("lifecycle seq=%d err=%v", seq, err)
65 }
66 // Sidecar create and typed decode/rewrite preserve the counter too.
67 side := provision.SidecarRecord{Schema: 1, Account: "bab", App: "app", Name: "worker", Host: provision.SidecarHost("worker", "app", "bab"), CreatedAt: f.now}
68 key = provision.SidecarRecordKey("bab", "app", "worker")
69 permitted := context.WithValue(ctx, effectPermitKey{}, effectPermit{Resource: key})
70 body, _ = json.Marshal(side)
71 tag, err := records.Put(permitted, key, body, "")
72 if err != nil {
73 t.Fatal(err)
74 }
75 body, _, _ = records.Get(ctx, key)
76 if err := json.Unmarshal(body, &side); err != nil || side.RecordSeq != 1 {
77 t.Fatalf("sidecar: %+v %v", side, err)
78 }
79 body, _ = json.Marshal(side)
80 if _, err := records.Put(permitted, key, body, tag); err != nil {
81 t.Fatal(err)
82 }
83 body, _, _ = records.Get(ctx, key)
84 _, seq, err = recordSequence(body)
85 if err != nil || seq != 2 {
86 t.Fatalf("sidecar seq=%d %v", seq, err)
87 }
88}
89
90func TestPlan9B8SequenceCannotBeRemovedLoweredOrChosen(t *testing.T) {
91 for _, mode := range []string{"ordinary", "effect", "control-object", "stream"} {
92 for _, counter := range []string{"", `,"record_seq":0`, `,"record_seq":1`, `,"record_seq":3`, `,"record_seq":null`, `,"record_seq":-1`, `,"record_seq":1.5`, `,"record_seq":18446744073709551616`, `,"Record_Seq":2`, `,"record_seq":2,"record_seq":2`} {
93 t.Run(mode+counter, func(t *testing.T) {
94 f := newMoveStartFixture(t, nil)
95 ctx := context.Background()
96 body, tag, _ := f.raw.Get(ctx, f.raw.recordKey)
97 var fields map[string]json.RawMessage
98 _ = json.Unmarshal(body, &fields)
99 fields["record_seq"] = json.RawMessage(`2`)
100 before, _ := json.Marshal(fields)
101 tag, err := f.raw.Put(ctx, f.raw.recordKey, before, tag)
102 if err != nil {
103 t.Fatal(err)
104 }
105 delete(fields, "record_seq")
106 body, _ = json.Marshal(fields)
107 candidate := append(append([]byte(nil), body[:len(body)-1]...), []byte(counter+"}")...)
108 records, err := NewConditionalRecordStore(f.raw, f.pub)
109 if err != nil {
110 t.Fatal(err)
111 }
112 switch mode {
113 case "effect", "stream":
114 ctx = context.WithValue(ctx, effectPermitKey{}, effectPermit{Resource: f.raw.recordKey})
115 case "control-object":
116 ctx = context.WithValue(ctx, effectPermitKey{}, effectPermit{Resource: "control-object:" + f.raw.recordKey})
117 }
118 if mode == "stream" {
119 _, err = release.PutObjectFrom(ctx, records, f.raw.recordKey, bytes.NewReader(candidate), int64(len(candidate)), tag)
120 } else {
121 _, err = records.Put(ctx, f.raw.recordKey, candidate, tag)
122 }
123 if !errors.Is(err, ErrIdentity) {
124 t.Fatalf("counter accepted: %s err=%v", candidate, err)
125 }
126 after, afterTag, _ := f.raw.Get(ctx, f.raw.recordKey)
127 if !bytes.Equal(before, after) || afterTag != tag {
128 t.Fatal("refusal changed record")
129 }
130 })
131 }
132 }
133}
134
135func TestPlan9B8SequencePrecisionOverflowAndUnknownFields(t *testing.T) {
136 const high = uint64(1<<53) + 1
137 before := []byte(fmt.Sprintf(`{"record_seq":%d,"future":{"value":9007199254740993}}`, high))
138 body, err := advanceRecordSequence(before, before)
139 fields, seq, parseErr := recordSequence(body)
140 if err != nil || parseErr != nil || seq != high+1 || string(fields["future"]) != `{"value":9007199254740993}` {
141 t.Fatalf("precision: %s %v %v", body, err, parseErr)
142 }
143 max := []byte(`{"record_seq":18446744073709551615}`)
144 if _, err := advanceRecordSequence(max, max); !errors.Is(err, ErrPending) {
145 t.Fatalf("overflow: %v", err)
146 }
147}
148
149func TestPlan9B8FlagOffBytesAndNonRecordBytesUnchanged(t *testing.T) {
150 f := newMoveStartFixture(t, nil)
151 ctx := context.Background()
152 before, tag, _ := f.raw.Get(ctx, f.raw.recordKey)
153 off, err := NewMutationBoundary(false, f.raw, nil)
154 if err != nil {
155 t.Fatal(err)
156 }
157 if err := provision.SetRecordRegion(ctx, off, "bab", "app", f.host, "ie"); err != nil {
158 t.Fatal(err)
159 }
160 after, got, _ := f.raw.Get(ctx, f.raw.recordKey)
161 if !bytes.Equal(before, after) || got != tag || bytes.Contains(after, []byte(`"record_seq"`)) {
162 t.Fatalf("flag-off changed: %s", after)
163 }
164 records, err := NewConditionalRecordStore(f.raw, f.pub)
165 if err != nil {
166 t.Fatal(err)
167 }
168 key := "apps/bab/app/metadata.json"
169 body := []byte("{\"value\": 1}\n")
170 if _, err := records.Put(ctx, key, body, ""); err != nil {
171 t.Fatal(err)
172 }
173 after, _, _ = records.Get(ctx, key)
174 if !bytes.Equal(body, after) {
175 t.Fatal("non-record bytes changed")
176 }
177 side, _ := json.Marshal(provision.SidecarRecord{Name: "worker"})
178 if strings.Contains(string(side), "record_seq") {
179 t.Fatalf("legacy sidecar: %s", side)
180 }
181}
182
183func TestPlan9B8ResolverRefusesMissingReplacedOrMalformedRecord(t *testing.T) {
184 for _, mode := range []string{"missing", "incarnation", "null-sequence", "lower-sequence", "uncaptured", "exhausted"} {
185 t.Run(mode, func(t *testing.T) {
186 ctx := context.Background()
187 r := newResolveFixture(t, false)
188 body, tag := r.record(t)
189 fields, _, _ := recordSequence(body)
190 switch mode {
191 case "missing":
192 if err := r.raw.Store.Delete(ctx, r.raw.recordKey); err != nil {
193 t.Fatal(err)
194 }
195 case "incarnation":
196 fields["created_at"] = json.RawMessage(`"2026-10-08T00:00:00Z"`)
197 case "null-sequence":
198 fields["record_seq"] = json.RawMessage(`null`)
199 case "lower-sequence":
200 r.h.moveStart.puts.Load().bases = []uint64{1}
201 case "uncaptured":
202 r.h.moveStart.puts.Load().bases = nil
203 case "exhausted":
204 fields["record_seq"] = json.RawMessage(`18446744073709551615`)
205 r.h.moveStart.puts.Load().bases = []uint64{^uint64(0)}
206 }
207 if mode == "incarnation" || mode == "null-sequence" || mode == "exhausted" {
208 next, _ := json.Marshal(fields)
209 if _, err := r.raw.Store.Put(ctx, r.raw.recordKey, next, tag); err != nil {
210 t.Fatal(err)
211 }
212 }
213 if got, err := ResolveMoveStart(ctx, r.h, r.moving, moveStartID); err == nil || got != OutcomeUncertain {
214 t.Fatalf("refusal: %q %v", got, err)
215 }
216 if r.effect(t).State != "uncertain" || r.control.fences != 0 {
217 t.Fatal("unproven resolution wrote or settled")
218 }
219 })
220 }
221}
222
223type sequenceFenceFailure struct {
224 release.Store
225 key string
226 land bool
227 conflicts bool
228 calls int
229}
230
231func (s *sequenceFenceFailure) Put(ctx context.Context, key string, body []byte, tag string) (string, error) {
232 if key != s.key {
233 return s.Store.Put(ctx, key, body, tag)
234 }
235 s.calls++
236 if s.conflicts {
237 return "", release.ErrCASConflict
238 }
239 if s.land {
240 if _, err := s.Store.Put(ctx, key, body, tag); err != nil {
241 return "", err
242 }
243 }
244 return "", errors.New("fence answer lost")
245}
246func TestPlan9B8FenceFailureRemainsUncertainAndCanRecover(t *testing.T) {
247 for _, land := range []bool{false, true} {
248 t.Run(fmt.Sprint(land), func(t *testing.T) {
249 ctx := context.Background()
250 r := newResolveFixture(t, false)
251 failure := &sequenceFenceFailure{Store: r.control, key: r.raw.recordKey, land: land}
252 r.h.w.raw = failure
253 if got, err := ResolveMoveStart(ctx, r.h, r.moving, moveStartID); err == nil || got != OutcomeUncertain {
254 t.Fatalf("lost fence: %q %v", got, err)
255 }
256 if r.effect(t).State != "uncertain" {
257 t.Fatal("lost fence settled")
258 }
259 r.h.w.raw = r.control
260 if got, err := ResolveMoveStart(ctx, r.h, r.moving, moveStartID); err != nil || got != OutcomeNotApplied {
261 t.Fatalf("retry: %q %v", got, err)
262 }
263 if err := r.latePut(); !errors.Is(err, release.ErrCASConflict) {
264 t.Fatalf("late Put: %v", err)
265 }
266 })
267 }
268 r := newResolveFixture(t, false)
269 failure := &sequenceFenceFailure{Store: r.control, key: r.raw.recordKey, conflicts: true}
270 r.h.w.raw = failure
271 if got, err := ResolveMoveStart(context.Background(), r.h, r.moving, moveStartID); err == nil || got != OutcomeUncertain || failure.calls != 3 {
272 t.Fatalf("bounded conflicts: %q %v calls=%d", got, err, failure.calls)
273 }
274 if r.effect(t).State != "uncertain" {
275 t.Fatal("conflicts settled")
276 }
277}
278
279func TestPlan9B8CapturesEveryCASAttemptBase(t *testing.T) {
280 ctx := context.Background()
281 r := newResolveFixture(t, false)
282 p := r.h.moveStart.puts.Load()
283 // Model updateRecord's retry reading another revision before its own Put.
284 body, tag := r.record(t)
285 next, err := advanceRecordSequence(body, body)
286 if err != nil {
287 t.Fatal(err)
288 }
289 if _, err := r.raw.Store.Put(ctx, r.raw.recordKey, next, tag); err != nil {
290 t.Fatal(err)
291 }
292 failed := &lostRecordPut{Store: r.raw, recordKey: r.raw.recordKey, armed: true}
293 recorder := moveStartRecorder{s: failed, puts: p}
294 current, etag, err := recorder.Get(ctx, r.raw.recordKey)
295 if err != nil {
296 t.Fatal(err)
297 }
298 if _, err := recorder.Put(ctx, r.raw.recordKey, current, etag); err == nil {
299 t.Fatal("lost retry succeeded")
300 }
301 if high, ok := p.highestBase(); !ok || high != 1 || len(p.bases) != 2 {
302 t.Fatalf("bases: %v", p.bases)
303 }
304 if got, err := ResolveMoveStart(ctx, r.h, r.moving, moveStartID); err != nil || got != OutcomeNotApplied || r.control.fences != 1 {
305 t.Fatalf("retry fence: %q %v fences=%d", got, err, r.control.fences)
306 }
307 if _, err := r.raw.Store.Put(ctx, r.raw.recordKey, failed.body, failed.etag); !errors.Is(err, release.ErrCASConflict) {
308 t.Fatalf("late retry: %v", err)
309 }
310}
311
312func TestPlan9B8TerminalJournalStillProvesComplete(t *testing.T) {
313 for _, outcome := range []string{"done", "rolled_back", "aborted", "different-id"} {
314 t.Run(outcome, func(t *testing.T) {
315 ctx := context.Background()
316 r := newResolveFixture(t, outcome != "different-id")
317 id := moveStartID
318 want := OutcomeComplete
319 if outcome == "different-id" {
320 id = "mv_" + strings.Repeat("b", 26)
321 want = OutcomeNotApplied
322 }
323 // Complete describes the record-start effect, including a move that
324 // subsequently aborted or rolled back. The journal retains its outcome.
325 _, err := provision.SetRecordMove(provision.WithRecordSequence(ctx), r.raw.Store, "bab", "app", r.host, func(rec *provision.Record) error {
326 rec.LastMove = &provision.MoveOutcome{ID: id, Outcome: outcome, At: r.now}
327 rec.Move = nil
328 return nil
329 })
330 if err != nil {
331 t.Fatal(err)
332 }
333 if got, err := ResolveMoveStart(ctx, r.h, r.moving, moveStartID); err != nil || got != want {
334 t.Fatalf("terminal evidence: %q %v want %q", got, err, want)
335 }
336 result := moveStartID
337 if want == OutcomeNotApplied {
338 result = "not-applied"
339 }
340 if r.control.fences != 0 || r.effect(t).ResultReference != result {
341 t.Fatal("wrong terminal resolution")
342 }
343 })
344 }
345}
346