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 "crypto/sha256"
6 "database/sql"
7 "encoding/hex"
8 "encoding/json"
9 "errors"
10 "fmt"
11 "os"
12 "path/filepath"
13 "strings"
14 "testing"
15
16 "github.com/carlosframework/platform/internal/hostbox"
17 "github.com/carlosframework/platform/internal/hostclaim"
18 "github.com/carlosframework/platform/internal/hostguard"
19 "github.com/carlosframework/platform/internal/hostwrite"
20 "github.com/carlosframework/platform/internal/provision"
21 "github.com/carlosframework/platform/internal/registry"
22 "github.com/carlosframework/platform/internal/rootstate"
23 "github.com/carlosframework/platform/internal/schedule"
24 "time"
25)
26
27func TestPlan14TimerIntentionBeforeMaterialization(t *testing.T) {
28 plan14TimerIntention(t, func(t *testing.T) rootstate.Store { return &rollbackDisk{dir: t.TempDir()} })
29}
30
31type timerSyncFault struct {
32 rootstate.Store
33 phase string
34}
35
36func (s *timerSyncFault) Write(name string, body []byte) error {
37 if err := s.Store.Write(name, body); err != nil {
38 return err
39 }
40 pending := strings.Contains(string(body), `"PendingInstallation"`)
41 if s.phase == "pending" && pending || s.phase == "complete" && !pending {
42 s.phase = ""
43 return os.ErrInvalid
44 }
45 return nil
46}
47
48func plan14TimerIntention(t *testing.T, open func(*testing.T) rootstate.Store) {
49 for _, attack := range []string{"none", "already-materialized", "late-intention", "changed-registration", "changed-row", "omit-intention", "crash-after-intention", "crash-after-row", "intention-sync", "materialization-sync", "replace", "remove", "replace-sync", "remove-sync", "replace-crash", "remove-crash", "replace-advanced-crash", "remove-advanced-crash", "producer", "bootstrap"} {
50 t.Run(attack, func(t *testing.T) {
51 t.Setenv("CARLOS_HOST_CLAIM", "1")
52 f := newRecoveryFixture(t)
53 o := f.read()
54 h := o.Claim.Holder
55 plan14PutRecord(t, f.claims, provision.Record{Schema: 1, Host: o.Host, Account: h.Account, App: h.App, Channel: "stable", CreatedAt: f.now})
56 _, etag, _ := f.claims.Get(t.Context(), h.Key)
57 journals, err := hostbox.NewJournalService(f.journal)
58 if err != nil {
59 t.Fatal(err)
60 }
61 registryPath := filepath.Join(t.TempDir(), "registry.db")
62 reg, err := registry.Open(registryPath)
63 if err != nil {
64 t.Fatal(err)
65 }
66 defer reg.Close()
67 sum := sha256.Sum256([]byte(fmt.Sprintf("%s\x00%s\x00%d\x00%s", o.Host, h.Incarnation, h.Stamp, "box-1")))
68 owner := registry.ClaimOwnership{ClaimIncarnation: h.Incarnation, ClaimStamp: h.Stamp, ClaimOwner: hex.EncodeToString(sum[:]), ClaimRowID: "route"}
69 var bootstrap *registry.Timer
70 if attack == "bootstrap" {
71 if err := reg.UpsertTimer(registry.Timer{Host: o.Host, Name: "daily", Kind: "schedule", Path: "/daily", Spec: "every 1h0m0s", Attempts: 2, NextAt: 17}); err != nil {
72 t.Fatal(err)
73 }
74 bootstrap, err = reg.GetTimer(o.Host, "daily")
75 if err != nil {
76 t.Fatal(err)
77 }
78 owned := owner
79 owned.ClaimRowID = bootstrap.Registration
80 if err := reg.AdoptClaimTimer(t.Context(), *bootstrap, owned); err != nil {
81 t.Fatal(err)
82 }
83 }
84 if err := reg.Upsert(registry.Route{Host: o.Host, Account: h.Account, App: h.App, DBPath: provision.NestedDBPath(f.guard.DataDir, o.Host), ClaimOwnership: owner}); err != nil {
85 t.Fatal(err)
86 }
87 in := hostbox.StopInventory{Complete: true, Host: o.Host, Incarnation: h.Incarnation, InstallationStamp: h.Stamp, Box: "box-1", Enrollment: "en", DirectoryRevision: "1", Entries: []hostbox.StopInventoryEntry{{Kind: "route", ArtifactID: "route", ManagedRoot: "registry", RelativePath: o.Host, RowIdentity: owner.ClaimRowID, Ownership: owner.ClaimOwner, Materialization: "materialized"}}}
88 artifacts := hostguard.InstallationArtifacts{Schema: 1, StateDirectory: true}
89 if bootstrap != nil {
90 artifacts.Timers = bootstrap.Name
91 in.Entries = append(in.Entries, hostbox.StopInventoryEntry{Kind: "timer", ArtifactID: bootstrap.Registration, ManagedRoot: "registry", RelativePath: o.Host + "/" + bootstrap.Name, RowIdentity: bootstrap.Registration, Ownership: owner.ClaimOwner, Materialization: "materialized"})
92 }
93 if err := journals.Inventory.WriteStopInventory(t.Context(), in); err != nil {
94 t.Fatal(err)
95 }
96 g := f.guard
97 state := &timerSyncFault{Store: open(t)}
98 defer state.Close()
99 g.Root = hostguard.NewRootAuthority(state)
100 if err := g.EstablishRoot(t.Context(), hostguard.RootInstallation{Artifacts: artifacts, Host: o.Host, Incarnation: h.Incarnation, Stamp: h.Stamp, Account: h.Account, App: h.App, Record: h.Key, Box: "box-1", Enrollment: "en", Migration: 1, InventoryRevision: 1}, func(*hostclaim.Claim) error { return nil }); err != nil {
101 t.Fatal(err)
102 }
103 actor := hostbox.Actor{Box: "box-1", Enrollment: "en", Revision: 1, Binding: "1"}
104 root := ClaimRootHTTP{Stop: ClaimStopHTTP{Coordinator: ClaimStopCoordinator{Guard: g, Evidence: f.claims}}, Journals: journals, InventoryActor: func(context.Context) (hostbox.Actor, error) { return actor, nil }, Roots: map[string]string{"data": g.DataDir}, Reconcile: hostbox.ReconcileAdapter{Registry: reg, Census: &hostbox.EmptyInstallationCensus{}}}
105 var last hostbox.JournalRequest
106 _, client := hostbox.RemoteJournals(func(ctx context.Context, q hostbox.JournalRequest) (hostbox.JournalResponse, error) {
107 raw, err := json.Marshal(q)
108 if err != nil {
109 t.Fatal(err)
110 }
111 q = hostbox.JournalRequest{}
112 if err := json.Unmarshal(raw, &q); err != nil {
113 t.Fatal(err)
114 }
115 if q.Op == "inventory-version" {
116 out, err := journals.Inventory.SnapshotRequest(ctx, q.Target)
117 out.V, out.ID = 1, q.ID
118 return out, err
119 }
120 last = q
121 return root.commitRootInventory(ctx, q)
122 })
123 expect := hostbox.InventoryExpectation{Revision: 1, RecordKey: h.Key, RecordETag: etag, Target: hostguard.FenceTarget{Host: o.Host, Incarnation: h.Incarnation, Stamp: h.Stamp, Session: h.Session, Box: actor.Box, Enrollment: actor.Enrollment, ClaimETag: o.ETag, Op: o.Claim.Op.ID, Phase: o.Claim.Op.Phase}}
124 write := func() error {
125 _, err := client.WriteExpected(t.Context(), in, expect)
126 if err == nil {
127 expect.Revision++
128 }
129 return err
130 }
131 timer := registry.Timer{Host: o.Host, Name: "daily", Kind: "schedule", Spec: "every 1h0m0s", Path: "/daily", Registration: "planned", ClaimOwnership: owner}
132 timer.ClaimRowID = timer.Registration
133 if attack == "producer" || attack == "bootstrap" {
134 serviceGuard := f.guard
135 if err := serviceGuard.Activate(t.Context(), o.Host, actor.Box, nil); err != nil {
136 t.Fatal(err)
137 }
138 serviceGuard.FenceView = g.Root.View
139 adapter := hostbox.ReconcileAdapter{Registry: reg, Inventory: client, Box: actor.Box, Enrollment: actor.Enrollment, DirectoryRevision: "1", Reconciler: hostbox.Reconciler{Guard: serviceGuard, Claims: f.claims}}
140 source := hostbox.WorkSource{Guard: serviceGuard, Box: actor.Box, Region: "ie", Now: time.Now, Policy: func(context.Context, *hostclaim.Claim, hostbox.Selection, hostwrite.Identity) error { return nil }, Local: func(context.Context, *hostclaim.Claim, hostbox.Selection) error { return nil }, Timer: adapter.MutateTimer}
141 work, err := source.Select(t.Context(), o.Host, "timers")
142 if err != nil {
143 t.Fatal(err)
144 }
145 producer := schedule.SelectedTimers{Store: reg, Work: work, Context: t.Context(), Host: o.Host}
146 timer.Registration = ""
147 timer.ClaimOwnership = registry.ClaimOwnership{}
148 if err := producer.UpsertTimer(timer); err != nil {
149 t.Fatal("producer registration", err)
150 }
151 first, err := reg.GetTimer(o.Host, timer.Name)
152 if err != nil {
153 t.Fatal(err)
154 }
155 if err := producer.UpsertTimer(timer); err != nil {
156 t.Fatal("producer replacement", err)
157 }
158 second, err := reg.GetTimer(o.Host, timer.Name)
159 if err != nil || first.Registration == second.Registration {
160 t.Fatal("producer identity", second, err)
161 }
162 if err := producer.DeleteTimer(o.Host, timer.Name); err != nil {
163 t.Fatal("producer withdrawal", err)
164 }
165 if _, err := reg.GetTimer(o.Host, timer.Name); !errors.Is(err, os.ErrNotExist) {
166 t.Fatal("producer deletion", err)
167 }
168 if err := producer.UpsertTimer(timer); err != nil {
169 t.Fatal("producer registration after withdrawal", err)
170 }
171 imported := timer
172 imported.Registration = "imported"
173 imported.Attempts, imported.LastAt, imported.NextAt, imported.CreatedAt = 3, 7, 11, 13
174 if err := producer.ReplaceTimers(o.Host, []registry.Timer{imported}); err != nil {
175 t.Fatal("import", err)
176 }
177 if err := producer.ReplaceTimers(o.Host, []registry.Timer{imported}); err != nil {
178 t.Fatal("import replay", err)
179 }
180 got, err := reg.GetTimer(o.Host, imported.Name)
181 if err != nil || got.Registration != "imported" || got.Attempts != 3 || got.LastAt != 7 || got.NextAt != 11 || got.CreatedAt != 13 {
182 t.Fatal("import identity and bookkeeping", got, err)
183 }
184 return
185 }
186
187 row := func() {
188 t.Helper()
189 if err := reg.ReplaceTimers(o.Host, []registry.Timer{timer}); err != nil {
190 t.Fatal(err)
191 }
192 }
193 in.Entries = append(in.Entries, hostbox.StopInventoryEntry{Kind: "timer", ArtifactID: "timer", ManagedRoot: "registry", RelativePath: o.Host + "/daily", RowIdentity: timer.Registration, Ownership: owner.ClaimOwner, Materialization: "intended"})
194 if attack == "already-materialized" || attack == "late-intention" {
195 row()
196 if attack == "already-materialized" {
197 in.Entries[1].Materialization = "materialized"
198 }
199 if err := write(); err == nil {
200 t.Fatal("adopted timer without prior root intention")
201 }
202 return
203 }
204 if attack == "intention-sync" {
205 state.phase = "pending"
206 if err := write(); err == nil {
207 t.Fatal("acknowledged unsynced intention")
208 }
209 if err := root.recoverRootInventory(t.Context()); err != nil {
210 t.Fatal("recover intended write", err)
211 }
212 if _, err := root.commitRootInventory(t.Context(), last); err != nil {
213 t.Fatal("replay recovered intention", err)
214 }
215 expect.Revision++
216 } else if err := write(); err != nil {
217 t.Fatal("record intention", err)
218 }
219 if attack == "crash-after-intention" {
220 if err := write(); err != nil {
221 t.Fatal("resume absent intended row", err)
222 }
223 }
224 row()
225 if attack == "crash-after-row" {
226 if err := write(); err != nil {
227 t.Fatal("resume intended exact row", err)
228 }
229 }
230 in.Entries[1].Materialization = "materialized"
231 if attack == "materialization-sync" {
232 state.phase = "complete"
233 }
234 switch attack {
235 case "changed-registration", "changed-row":
236 db, err := sql.Open("sqlite", registryPath)
237 if err != nil {
238 t.Fatal(err)
239 }
240 column := "registration"
241 if attack == "changed-row" {
242 column = "claim_row_id"
243 }
244 _, err = db.Exec("UPDATE timers SET "+column+"='successor' WHERE host=? AND name=?", o.Host, timer.Name)
245 db.Close()
246 if err != nil {
247 t.Fatal(err)
248 }
249 case "omit-intention":
250 in.Entries = in.Entries[:1]
251 }
252 err = write()
253 if attack == "materialization-sync" {
254 if err == nil {
255 t.Fatal("acknowledged unsynced materialization")
256 }
257 _, err = root.commitRootInventory(t.Context(), last)
258 if err == nil {
259 expect.Revision++
260 }
261 }
262 valid := attack == "none" || attack == "crash-after-intention" || attack == "crash-after-row" || attack == "intention-sync" || attack == "materialization-sync" || strings.HasPrefix(attack, "replace") || strings.HasPrefix(attack, "remove")
263 if (err == nil) != valid {
264 t.Fatalf("materialization valid=%t: %v", valid, err)
265 }
266 if valid {
267 if err := write(); err != nil {
268 t.Fatal("materialized replay", err)
269 }
270 if strings.HasPrefix(attack, "replace") || strings.HasPrefix(attack, "remove") {
271 mutation := hostbox.TimerMutation{Timer: timer, Expected: timer.Registration, Remove: strings.HasPrefix(attack, "remove")}
272 if mutation.Remove {
273 in.Entries[1].TimerRetired = true
274 } else {
275 mutation.Timer.Registration = "replacement"
276 mutation.Timer.ClaimRowID = "replacement"
277 in.Entries[1].RowIdentity = "replacement"
278 }
279 if strings.HasSuffix(attack, "sync") {
280 state.phase = "complete"
281 }
282 if strings.HasSuffix(attack, "crash") {
283 calls := 0
284 journals.Now = func() time.Time {
285 calls++
286 if calls > 1 {
287 return time.Now().Add(2 * hostbox.JournalRequestWindow)
288 }
289 return time.Now()
290 }
291 }
292 revision, err := client.WriteTimerExpected(t.Context(), in, expect, mutation)
293 if strings.HasSuffix(attack, "crash") {
294 if err == nil {
295 t.Fatal("journal fault did not interrupt mutation")
296 }
297 got, getErr := reg.GetTimer(timer.Host, timer.Name)
298 if mutation.Remove {
299 if !errors.Is(getErr, os.ErrNotExist) {
300 t.Fatal("did not reach row removal", getErr)
301 }
302 } else if getErr != nil || got.Registration != mutation.Timer.Registration {
303 t.Fatal("did not reach row replacement", got, getErr)
304 }
305 journals.Now = time.Now
306 if strings.Contains(attack, "advanced") {
307 key, _ := hostclaim.Key(o.Host)
308 body, etag, err := f.claims.Get(t.Context(), key)
309 if err != nil {
310 t.Fatal(err)
311 }
312 if _, err := f.claims.Put(t.Context(), key, append(body, '\n'), etag); err != nil {
313 t.Fatal(err)
314 }
315 }
316 if err := root.recoverRootInventory(t.Context()); err != nil {
317 t.Fatal(err)
318 }
319 if strings.Contains(attack, "advanced") {
320 snapshot, err := journals.Inventory.SnapshotRequest(t.Context(), expect.Target)
321 if err != nil || snapshot.Revision != expect.Revision+1 || snapshot.Inventory.Entries[1] != in.Entries[1] {
322 t.Fatal("advanced claim lost pending effect", snapshot, err)
323 }
324 pending, err := g.Root.PendingInstallations()
325 if err != nil || len(pending) != 0 {
326 t.Fatal("effect still pending", pending, err)
327 }
328 return
329 }
330 reply, replayErr := root.commitRootInventory(t.Context(), last)
331 err, revision = replayErr, reply.Revision
332 }
333
334 if strings.HasSuffix(attack, "sync") {
335 if err == nil {
336 t.Fatal("acknowledged failed sync")
337 }
338 reply, replayErr := root.commitRootInventory(t.Context(), last)
339 err, revision = replayErr, reply.Revision
340 }
341 if err != nil || revision != expect.Revision+1 {
342 t.Fatal("timer mutation", revision, err)
343 }
344 expect.Revision = revision
345 got, err := reg.GetTimer(o.Host, timer.Name)
346 if mutation.Remove {
347 if !errors.Is(err, os.ErrNotExist) {
348 t.Fatal("timer survived", got, err)
349 }
350 } else if err != nil || got.Registration != "replacement" {
351 t.Fatal("replacement missing", got, err)
352 }
353 if _, err := root.commitRootInventory(t.Context(), last); err != nil {
354 t.Fatal("mutation replay", err)
355 }
356 }
357
358 got, err := journals.Inventory.ReadStopInventory(t.Context(), o.Host)
359 if err != nil || len(got.Entries) != 2 || got.Entries[1] != in.Entries[1] {
360 t.Fatal("lost exact timer", got, err)
361 }
362 }
363 })
364 }
365}
366