amadan

carlos / platform Public

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

Plain git — no account needed to clone.

Download

Download this file

1package release
2
3import (
4 "context"
5 "crypto/ed25519"
6 "errors"
7 "testing"
8 "time"
9
10 "github.com/carlosframework/platform/internal/registry"
11)
12
13// Settings plan 4, Astra round 2 finding 2 (ruling R-CARRY-FENCE): a root
14// pass already running when a graceful cut begins publishes its request
15// record before the cut's carry, and is carried, or not at all. A late
16// record is never left behind on the source box.
17//
18// The root pass is the real one (Pass with the update settings, root's
19// resolve). It is paused by the test hook at one of the two points of its
20// fenced publication. The cut's half is the agent's carry as
21// buildStates.carryRequest runs it: the row held moving first, then the
22// fence (FenceCarry), the consumed file read and merged (CarryRequest), the
23// fence released.
24
25const fenceHost = "a.example.com"
26
27// fenceCutResult is what the cut sealed: the build state after its carry.
28type fenceCutResult struct {
29 state BuildState
30 err error
31}
32
33// fenceCut runs the cut's half: marked is closed once the row is held.
34func fenceCut(reg *registry.Registry, opt string, state BuildState, marked chan<- struct{}) fenceCutResult {
35 if err := reg.SetStatus(fenceHost, "moving"); err != nil {
36 return fenceCutResult{err: err}
37 }
38 close(marked)
39 unlock, err := FenceCarry(context.Background(), opt, fenceHost)
40 if err != nil {
41 return fenceCutResult{err: err}
42 }
43 defer unlock()
44 c, err := ReadConsumed(opt, fenceHost)
45 if err != nil {
46 return fenceCutResult{err: err}
47 }
48 return fenceCutResult{state: CarryRequest(&state, c)}
49}
50
51// fenceScenario is one root pass over one route: setup prepares the store,
52// registry and consumed file and returns the build state the cut would
53// seal and the pass options.
54type fenceScenario struct {
55 name string
56 setup func(t *testing.T, s Store, priv ed25519.PrivateKey, opt string, reg *registry.Registry) (BuildState, []PassOption)
57 // published reports whether the consumed file holds the pass's record.
58 published func(c *Consumed) bool
59}
60
61func fenceScenarios() []fenceScenario {
62 ctx := context.Background()
63 return []fenceScenario{
64 {
65 // Update now, resolved by the pass to the head before staging.
66 name: "a resolution",
67 setup: func(t *testing.T, s Store, priv ed25519.PrivateKey, opt string, reg *registry.Registry) (BuildState, []PassOption) {
68 shipRelease(t, s, "bab", "km", "v1", adoptPlatform, []byte("v1"))
69 if err := Promote(ctx, s, priv, "bab", "km", "canary/x", "v1"); err != nil {
70 t.Fatal(err)
71 }
72 st := BuildState{LastExecuted: "v0"}
73 settingsRoute(t, reg, fenceHost, "km", "canary/x", UpdateSettings{Updates: UpdatesManual, Channel: "canary/x", InstallRequest: 1},
74 BuildStateCopy{Known: true, Replica: true, State: &st})
75 return st, nil
76 },
77 published: func(c *Consumed) bool { return c != nil && c.Resolved != nil && c.Resolved.ID == 1 },
78 },
79 {
80 // R1 resolved to v2, which the failure book holds: the pass
81 // records R1 completed, failed.
82 name: "a failed completion",
83 setup: func(t *testing.T, s Store, priv ed25519.PrivateKey, opt string, reg *registry.Registry) (BuildState, []PassOption) {
84 for _, v := range []string{"v1", "v2"} {
85 shipRelease(t, s, "bab", "km", v, adoptPlatform, []byte(v))
86 if err := Promote(ctx, s, priv, "bab", "km", "canary/x", v); err != nil {
87 t.Fatal(err)
88 }
89 }
90 st := BuildState{LastExecuted: "v1", Resolved: &ResolvedRequest{ID: 1, Target: "v2"}}
91 settingsRoute(t, reg, fenceHost, "km", "canary/x", UpdateSettings{Updates: UpdatesManual, Channel: "canary/x", InstallRequest: 1},
92 BuildStateCopy{Known: true, Replica: true, State: &st})
93 if err := WriteConsumed(opt, Consumed{Host: fenceHost, Resolved: &ResolvedRequest{ID: 1, Target: "v2"}}); err != nil {
94 t.Fatal(err)
95 }
96 book := func() (FailedBuild, error) {
97 return func(_, _, build, _ string, _ bool) bool { return build == "v2" }, nil
98 }
99 return st, []PassOption{WithUpdateSettings(book, func(registry.Route) bool { return false }, true)}
100 },
101 published: func(c *Consumed) bool { return c != nil && c.Completed == 1 && c.CompletedFailed },
102 },
103 }
104}
105
106func withFenceHook(t *testing.T, hook func(host, at string)) {
107 t.Helper()
108 old := beforeConsumedWrite
109 beforeConsumedWrite = hook
110 t.Cleanup(func() { beforeConsumedWrite = old })
111}
112
113// leftBehind reports whether the source's consumed file holds request state
114// the sealed build state lacks: carrying it again would change the state.
115func leftBehind(t *testing.T, opt string, sealed BuildState) bool {
116 t.Helper()
117 c, err := ReadConsumed(opt, fenceHost)
118 if err != nil {
119 t.Fatal(err)
120 }
121 again := CarryRequest(&sealed, c)
122 return !SameRequestState(&sealed, &again)
123}
124
125// A cut that holds the row and carries while root's pass is paused before
126// its fence: root then takes the fence, sees the hold, publishes nothing and
127// installs nothing. The record is never published, so nothing is left behind.
128func TestARootPassPausedBeforeTheFenceDoesNotPublishOverACut(t *testing.T) {
129 for _, sc := range fenceScenarios() {
130 t.Run(sc.name, func(t *testing.T) {
131 s := NewDirStore(t.TempDir())
132 priv, pub := testKeys(t)
133 opt, reg := t.TempDir(), testRegistry(t)
134 state, opts := sc.setup(t, s, priv, opt, reg)
135 before, _ := ReadConsumed(opt, fenceHost)
136 var cut fenceCutResult
137 withFenceHook(t, func(host, at string) {
138 if at == "fence" {
139 cut = fenceCut(reg, opt, state, make(chan struct{}))
140 }
141 })
142 passWithSettings(t, reg, s, pub, opt, opts...)
143 if cut.err != nil {
144 t.Fatalf("the cut: %v", cut.err)
145 }
146 after, _ := ReadConsumed(opt, fenceHost)
147 if sc.published(after) && !sc.published(before) {
148 t.Fatalf("root published %+v after the cut held the row and carried", after)
149 }
150 if leftBehind(t, opt, cut.state) {
151 t.Fatalf("a record was left behind: consumed %+v, sealed %+v", after, cut.state)
152 }
153 if v, _ := LiveVersion(opt, fenceHost); v != "" {
154 t.Fatalf("live %q: root installed for a route the cut holds", v)
155 }
156 })
157 }
158}
159
160// A root pass paused immediately before WriteConsumed, its fence held and
161// its re-check passed, while a cut holds the row and asks for the fence:
162// the cut waits, root's record lands, and the cut carries it.
163func TestARootPassPausedBeforeItsWriteIsCarriedByTheCut(t *testing.T) {
164 for _, sc := range fenceScenarios() {
165 t.Run(sc.name, func(t *testing.T) {
166 s := NewDirStore(t.TempDir())
167 priv, pub := testKeys(t)
168 opt, reg := t.TempDir(), testRegistry(t)
169 state, opts := sc.setup(t, s, priv, opt, reg)
170 done := make(chan fenceCutResult, 1)
171 early := false
172 withFenceHook(t, func(host, at string) {
173 if at != "write" {
174 return
175 }
176 marked := make(chan struct{})
177 go func() { done <- fenceCut(reg, opt, state, marked) }()
178 <-marked
179 select {
180 case <-done:
181 early = true
182 case <-time.After(200 * time.Millisecond):
183 }
184 })
185 passWithSettings(t, reg, s, pub, opt, opts...)
186 if early {
187 t.Fatal("the cut carried while root held the fence")
188 }
189 var cut fenceCutResult
190 select {
191 case cut = <-done:
192 case <-time.After(10 * time.Second):
193 t.Fatal("the cut never took the fence")
194 }
195 if cut.err != nil {
196 t.Fatalf("the cut: %v", cut.err)
197 }
198 after, _ := ReadConsumed(opt, fenceHost)
199 if !sc.published(after) {
200 t.Fatalf("consumed %+v: root's record did not land", after)
201 }
202 if leftBehind(t, opt, cut.state) {
203 t.Fatalf("a record was left behind: consumed %+v, sealed %+v", after, cut.state)
204 }
205 })
206 }
207}
208
209// With no move in progress the fenced pass is the pass it was: the hook runs
210// at both points and the record lands as without it.
211func TestARootPassWithNoMoveIsUnchangedByTheFence(t *testing.T) {
212 for _, sc := range fenceScenarios() {
213 t.Run(sc.name, func(t *testing.T) {
214 var want *Consumed
215 var wantLive string
216 for _, hooked := range []bool{false, true} {
217 s := NewDirStore(t.TempDir())
218 priv, pub := testKeys(t)
219 opt, reg := t.TempDir(), testRegistry(t)
220 _, opts := sc.setup(t, s, priv, opt, reg)
221 var seen []string
222 if hooked {
223 withFenceHook(t, func(_, at string) { seen = append(seen, at) })
224 }
225 passWithSettings(t, reg, s, pub, opt, opts...)
226 got, err := ReadConsumed(opt, fenceHost)
227 if err != nil || !sc.published(got) {
228 t.Fatalf("hooked %v: consumed %+v, %v; want the pass's record", hooked, got, err)
229 }
230 live, _ := LiveVersion(opt, fenceHost)
231 if !hooked {
232 want, wantLive = got, live
233 continue
234 }
235 if len(seen) != 2 || seen[0] != "fence" || seen[1] != "write" {
236 t.Fatalf("hook points %v, want fence then write", seen)
237 }
238 g, w := *got, *want
239 g.At, w.At = time.Time{}, time.Time{}
240 if !sameConsumedForTest(g, w) || live != wantLive {
241 t.Fatalf("fenced pass: consumed %+v live %q; unfenced: %+v live %q", g, live, w, wantLive)
242 }
243 }
244 })
245 }
246}
247
248func sameConsumedForTest(a, b Consumed) bool {
249 ra, rb := a.Resolved, b.Resolved
250 a.Resolved, b.Resolved = nil, nil
251 return a == b && ((ra == nil && rb == nil) || (ra != nil && rb != nil && *ra == *rb))
252}
253
254// publishGate: a held or missing row is refused; a ready one is not.
255func TestPublishGateRefusesAMoveHold(t *testing.T) {
256 for _, st := range []string{"moving", "landing", "restoring", "releasing"} {
257 if err := publishGate(&registry.Route{Host: fenceHost, Status: st}); !errors.Is(err, ErrMoveHeld) {
258 t.Errorf("status %q: %v; want ErrMoveHeld", st, err)
259 }
260 }
261 if err := publishGate(nil); !errors.Is(err, ErrMoveHeld) {
262 t.Errorf("no row: %v; want ErrMoveHeld", err)
263 }
264 for _, st := range []string{"", "ready"} {
265 if err := publishGate(&registry.Route{Host: fenceHost, Status: st}); err != nil {
266 t.Errorf("status %q: %v; want nil", st, err)
267 }
268 }
269}
270