amadan

carlos / platform Public

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

Plain git — no account needed to clone.

Download

Download this file

1package host
2
3import (
4 "context"
5 "errors"
6 "slices"
7 "strings"
8 "testing"
9
10 "github.com/carlosframework/platform/internal/activator"
11 "github.com/carlosframework/platform/internal/genselect"
12 "github.com/carlosframework/platform/internal/hostbox"
13 "github.com/carlosframework/platform/internal/hostclaim"
14 execctl "github.com/carlosframework/platform/internal/hostexec"
15 "github.com/carlosframework/platform/internal/hostrelocate"
16)
17
18// fakeReleaseReplica is the session's replicator as releaseFinalTXID sees
19// it: SyncAndWait.
20type fakeReleaseReplica struct {
21 calls []string
22 syncErr error
23}
24
25func (r *fakeReleaseReplica) SyncAndWait(context.Context) error {
26 r.calls = append(r.calls, "sync")
27 return r.syncErr
28}
29
30// Fix round 1, I2 (N1): an empty listing is not evidence of an empty
31// database. FinalTXID 0 comes only from writer session 0, a holder no
32// writer was ever admitted to; an empty listing for any other session (a
33// wrong bucket, a listing not yet visible) refuses. The listing is read
34// only after the replicator synced.
35func TestReleaseFinalTXIDIsZeroOnlyForWriterSessionZero(t *testing.T) {
36 ctx := context.Background()
37 failure := errors.New("s3: access denied")
38 for _, tc := range []struct {
39 name string
40 session uint64
41 rep fakeReleaseReplica
42 plan *genselect.Plan
43 selErr error
44 want uint64
45 refuses bool
46 calls []string
47 }{
48 {name: "the wrong bucket: an empty listing for writer session 6", session: 6, selErr: genselect.ErrEmpty, refuses: true},
49 {name: "a genuinely empty session-0 database", selErr: genselect.ErrEmpty, want: 0},
50 {name: "a listing at the database's position", session: 6, plan: &genselect.Plan{MaxTXID: 41, MaxLevel0TXID: 41}, want: 41},
51 {name: "level 0 behind the database", session: 6, plan: &genselect.Plan{MaxTXID: 41, MaxLevel0TXID: 40}, refuses: true},
52 {name: "a listing that fails", session: 6, selErr: failure, refuses: true},
53 {name: "a sync that fails", session: 0, rep: fakeReleaseReplica{syncErr: failure}, refuses: true, calls: []string{"sync"}},
54 } {
55 t.Run(tc.name, func(t *testing.T) {
56 rep := tc.rep
57 selected := false
58 sel := func(_ context.Context, r genselect.Request) (*genselect.Plan, error) {
59 selected = true
60 if r.Holder.Session != tc.session || r.Host != "shop.example" {
61 t.Fatalf("selected for %+v", r)
62 }
63 return tc.plan, tc.selErr
64 }
65 got, err := releaseFinalTXID(ctx, &rep, sel, genselect.Request{Host: "shop.example", Holder: hostclaim.Holder{Session: tc.session}})
66 if tc.refuses {
67 if err == nil {
68 t.Fatalf("released at %d", got)
69 }
70 } else if err != nil || got != tc.want {
71 t.Fatalf("got %d %v, want %d", got, err, tc.want)
72 }
73 if !slices.Equal(rep.calls, []string{"sync"}) || selected != (tc.calls == nil) {
74 t.Fatalf("ran %v selected=%v", rep.calls, selected)
75 }
76 })
77 }
78}
79
80// relocationFlushDB has nothing to flush without the session's replicator:
81// no wake, no attempt for the host, no data bucket.
82func TestRelocationFlushDBNeedsTheSessionsReplicator(t *testing.T) {
83 obs := hostclaim.Observed{Host: "shop.example", Claim: &hostclaim.Claim{Host: "shop.example", Holder: &hostclaim.Holder{Session: 0}}}
84 cfg := RelocationAssemblyConfig{Bucket: "data"}
85 for name, a := range map[string]ClaimAssembly{"no wake": {}, "no attempt": {Wake: &ClaimWakeAssembly{}}} {
86 if _, err := a.relocationFlushDB(cfg)(context.Background(), obs); !errors.Is(err, ErrClaimStopUnavailable) {
87 t.Fatalf("%s: %v", name, err)
88 }
89 }
90 if _, err := (ClaimAssembly{Wake: &ClaimWakeAssembly{}}).relocationFlushDB(cfg)(context.Background(), hostclaim.Observed{Host: "shop.example"}); !errors.Is(err, ErrClaimStopUnavailable) {
91 t.Fatalf("no claim: %v", err)
92 }
93}
94
95// relocationRoot is root's isolated launcher as relocationPreflight asks
96// it: the protocol, then the capability for one kind.
97type relocationRoot struct {
98 protocol string
99 kindErr error
100 asked []string
101}
102
103func (r *relocationRoot) Capability(context.Context) (string, error) {
104 r.asked = append(r.asked, "protocol")
105 return r.protocol, nil
106}
107func (*relocationRoot) Launch(context.Context, hostbox.RoleLaunchRequest) (hostbox.RoleLaunchGrant, error) {
108 return hostbox.RoleLaunchGrant{}, errors.New("unused")
109}
110func (*relocationRoot) Stop(context.Context, hostbox.RoleLaunchRequest) error { return nil }
111func (r *relocationRoot) KindCapability(_ context.Context, kind string) error {
112 r.asked = append(r.asked, "kind "+kind)
113 return r.kindErr
114}
115
116// Fix round 1, M4: relocationPreflight (ruling 48) asks root for the
117// isolated launch protocol, then for kind relocating; either refusal is
118// the answer. After P's opening, until claim ask C11, a later attempt is
119// ErrLaterAttemptUnlaunchable without asking root (ruling 58).
120func TestRelocationPreflightAsksRootForTheRelocatingKind(t *testing.T) {
121 ctx := context.Background()
122 refused := errors.New("root: no attempt reader for relocating")
123 probing := func(opened bool) hostclaim.Observed {
124 j := map[string]string{"prepared": "P1", "prepared_authority": `{"generation":"P1"}`, "pin": `{"empty":true}`, "probing_since": "2026-10-07T02:00:00Z"}
125 if opened {
126 j["prepared_opened"] = "P1"
127 }
128 return hostclaim.Observed{Host: "shop.example", Claim: &hostclaim.Claim{Host: "shop.example", Holder: &hostclaim.Holder{Session: 7},
129 Op: hostclaim.Op{Kind: hostclaim.KindRelocating, ID: "rel-1", Phase: hostclaim.PhaseRelocatingProbing, Journal: j}}}
130 }
131 for _, tc := range []struct {
132 name string
133 root *relocationRoot
134 opened bool
135 later bool
136 want error
137 asked []string
138 }{
139 {name: "root answers both", root: &relocationRoot{protocol: hostbox.RoleLaunchProtocol}, asked: []string{"protocol", "kind relocating"}},
140 {name: "root speaks another protocol", root: &relocationRoot{protocol: "old"}, want: activator.ErrClaimIsolationUnavailable, asked: []string{"protocol"}},
141 {name: "root refuses the kind", root: &relocationRoot{protocol: hostbox.RoleLaunchProtocol, kindErr: refused}, want: refused, asked: []string{"protocol", "kind relocating"}},
142 {name: "a later attempt before C11", root: &relocationRoot{protocol: hostbox.RoleLaunchProtocol}, opened: true, want: hostrelocate.ErrLaterAttemptUnlaunchable},
143 {name: "a later attempt once C11 is answered", root: &relocationRoot{protocol: hostbox.RoleLaunchProtocol}, opened: true, later: true, asked: []string{"protocol", "kind relocating"}},
144 } {
145 t.Run(tc.name, func(t *testing.T) {
146 a := ClaimAssembly{Wake: &ClaimWakeAssembly{Roles: &ClaimRoleAssembly{Root: tc.root}}}
147 err := a.relocationPreflight(RelocationAssemblyConfig{LaterProbeAttempts: tc.later})(ctx, probing(tc.opened))
148 if (tc.want == nil && err != nil) || (tc.want != nil && !errors.Is(err, tc.want)) {
149 t.Fatalf("err = %v, want %v", err, tc.want)
150 }
151 if !slices.Equal(tc.root.asked, tc.asked) {
152 t.Fatalf("asked root %v, want %v", tc.root.asked, tc.asked)
153 }
154 })
155 }
156 if err := (ClaimAssembly{}).relocationPreflight(RelocationAssemblyConfig{})(ctx, probing(false)); !errors.Is(err, activator.ErrClaimIsolationUnavailable) {
157 t.Fatalf("no roles: %v", err)
158 }
159}
160
161// Fix round 1, M4 (D5d): Acknowledge's one binding check. Main's
162// controller must be for exactly the observed op, incarnation and host;
163// any other is ErrLost, never an acknowledgement.
164func TestRelocationAcknowledgeBindsTheObservedOperation(t *testing.T) {
165 ctx := context.Background()
166 e, obs, _, _ := effectsFixture(t)
167 ack, err := e.Acknowledge(ctx, obs)
168 if err != nil || ack.Op != "rel-1" || ack.ControllerID != "c9" || ack.LeaseETag != "lease-1" || ack.Enrollment != "enr" || ack.BootID != "boot" {
169 t.Fatalf("the bound acknowledgement: %+v %v", ack, err)
170 }
171 for name, change := range map[string]func(*execctl.State){
172 "another op": func(s *execctl.State) { s.Op = "rel-0" },
173 "another incarnation": func(s *execctl.State) { s.Incarnation = "inc-0" },
174 "another host": func(s *execctl.State) { s.Host = "other.example" },
175 } {
176 e, obs, _, _ := effectsFixture(t)
177 inner := e.Participation
178 e.Participation = func(ctx context.Context, o hostclaim.Observed) (execctl.State, error) {
179 st, err := inner(ctx, o)
180 change(&st)
181 return st, err
182 }
183 if ack, err := e.Acknowledge(ctx, obs); !errors.Is(err, execctl.ErrLost) {
184 t.Fatalf("%s: acknowledged %+v %v", name, ack, err)
185 }
186 if err := e.Ready(ctx, obs); !errors.Is(err, execctl.ErrLost) {
187 t.Fatalf("%s: ready %v", name, err)
188 }
189 }
190}
191
192// Fix round 1, M4: validOpID is the one op-id check for a name under the
193// data root. Release positions refuse every id it refuses.
194func TestValidOpIDRefusesEveryPathThatIsNotOneName(t *testing.T) {
195 for _, op := range []string{"", ".", "..", "a/b", `a\b`, "a\x00b", "../rel-1", "rel-1/.."} {
196 if validOpID(op) {
197 t.Errorf("%q accepted", op)
198 }
199 f := ReleasePositionsFile{Dir: t.TempDir()}
200 r := StopRequest{Host: "shop.example", OperationID: op}
201 if err := f.Record(context.Background(), ReleasePosition{Request: r, PID: 1, ProcessStart: "1"}); !errors.Is(err, ErrClaimStopUnavailable) {
202 t.Errorf("%q recorded: %v", op, err)
203 }
204 if _, _, err := f.Recorded(context.Background(), r); !errors.Is(err, ErrClaimStopUnavailable) {
205 t.Errorf("%q read: %v", op, err)
206 }
207 }
208 for _, op := range []string{"rel-1", ".rel", "rel..1", "a.b"} {
209 if !validOpID(op) {
210 t.Errorf("%q refused", op)
211 }
212 }
213}
214
215// Fix round 1, M4: the release's Collect provider is a recovery request
216// (StopRequest.Recover) exactly when the release offers the recovery
217// drain, after a restart; the process that flushed asks with its own
218// producers and no Recover.
219func TestRelocationCollectRecoversAfterARestart(t *testing.T) {
220 ctx := context.Background()
221 e, obs, _, _ := effectsFixture(t)
222 var asked []StopRequest
223 a := ClaimAssembly{collectStop: func(ctx context.Context, r StopRequest, d StopDrainer) (StopResult, error) {
224 asked = append(asked, r)
225 if !r.Recover {
226 if _, err := d.DrainClaim(ctx, r, "fence"); err != nil {
227 return StopResult{}, err
228 }
229 }
230 return StopResult{Key: "control/hosts/shop.example/evidence/rel-1/src/r.json", Digest: strings.Repeat("a", 64)}, nil
231 }}
232 e.Collect = a.RelocationEffects(RelocationAssemblyConfig{}, nil).Collect
233 if _, err := e.Release(ctx, obs); err != nil {
234 t.Fatal(err)
235 }
236 e.Identity = func(context.Context) (RelocationIdentity, error) {
237 return RelocationIdentity{Actor: hostbox.Actor{Box: "src", Enrollment: "enr", Revision: 4, Binding: "4"}, BootID: "boot-2", ControllerID: "ctl-2", PID: 200, ProcessStart: "9000"}, nil
238 }
239 if _, err := e.Release(ctx, obs); err != nil {
240 t.Fatal(err)
241 }
242 if len(asked) != 2 || asked[0].Recover || !asked[1].Recover || asked[1].BootID != "boot" || asked[1].ControllerID != "ctl" {
243 t.Fatalf("asked root %+v", asked)
244 }
245}
246