amadan

carlos / platform Public

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

Plain git — no account needed to clone.

Download

Download this file

1//go:build linux && rootrunner
2
3package host
4
5import (
6 "bytes"
7 "context"
8 "crypto/ed25519"
9 "encoding/json"
10 "fmt"
11 "net"
12 "os"
13 "os/exec"
14 "path/filepath"
15 "strings"
16 "syscall"
17 "testing"
18 "time"
19
20 "github.com/carlosframework/platform/internal/activator"
21 "github.com/carlosframework/platform/internal/hostbox"
22 "github.com/carlosframework/platform/internal/hostclaim"
23 "github.com/carlosframework/platform/internal/hostfiles"
24 "github.com/carlosframework/platform/internal/hostguard"
25 "github.com/carlosframework/platform/internal/lease"
26 "github.com/carlosframework/platform/internal/lease/leasetest"
27 "github.com/carlosframework/platform/internal/registry"
28 "github.com/carlosframework/platform/internal/release"
29 "github.com/carlosframework/platform/internal/rootstate"
30)
31
32type seedCrashFixture struct {
33 Claims, Data, Locks, Root, Parks, Journal, Registry, Retirement, Signal string
34 Actor hostbox.Actor
35 Now time.Time
36}
37
38func rootFilesSeedWitnessFor(t *testing.T, binary string) filesSeedLiveWitness {
39 if strings.Contains(t.Name(), "Socket") {
40 return filesSeedLiveWitness{seed: rootFilesSeedLifetime}
41 }
42 if strings.Contains(t.Name(), "WholeExecutorKilledRerun") {
43 return filesSeedLiveWitness{seed: func(t *testing.T, p *servingPreparation, e *filesSeedExecutor) { rootFilesSeedCrash(t, binary, p, e) }}
44 }
45 return rootFilesSeedLiveWitness(t, binary)
46}
47
48func rootFilesSeedCrash(t *testing.T, binary string, p *servingPreparation, e *filesSeedExecutor) {
49 f := p.f
50 root := t.TempDir()
51 saved := seedCrashFixture{Claims: filepath.Join(root, "claims"), Data: f.f.guard.DataDir, Locks: f.f.guard.LockDir, Root: f.rootDisk.dir, Parks: "/var/lib/carlos-claims/authority/serving-parks", Journal: f.f.journalPath, Registry: f.p.root.Reconcile.Registry.Path(), Retirement: f.p.root.Reconcile.Retirement.JournalDir, Signal: filepath.Join(root, "precommit"), Actor: f.actor, Now: f.f.now}
52 claims := release.NewDirStore(saved.Claims)
53 keys, err := f.f.claims.List(t.Context(), "")
54 if err != nil {
55 t.Fatal(err)
56 }
57 for _, key := range keys {
58 body, _, err := f.f.claims.Get(t.Context(), key)
59 if err != nil {
60 t.Fatal(err)
61 }
62 if _, err = claims.Put(t.Context(), key, body, ""); err != nil {
63 t.Fatal(err)
64 }
65 }
66 // The seed selection registry and the handover registry share the same route here.
67 rt, err := e.registry.Get(recoveryHost)
68 if err != nil || rt == nil {
69 t.Fatal(err)
70 }
71 savedBody, _ := json.Marshal(saved)
72 config := filepath.Join(root, "fixture.json")
73 if err := os.WriteFile(config, savedBody, 0600); err != nil {
74 t.Fatal(err)
75 }
76 run := func(crash bool) error {
77 cmd := exec.Command(binary, "-test.run=^TestRootFilesSeedWholeExecutorPayload$", "-test.v", "-test.timeout=40s")
78 cmd.Env = append(os.Environ(), "FILES_SEED_CRASH_CONFIG="+config, fmt.Sprintf("FILES_SEED_CRASH=%v", crash))
79 cmd.Stdin = os.Stdin
80 var out bytes.Buffer
81 cmd.Stdout, cmd.Stderr = &out, &out
82 err := cmd.Run()
83 t.Log(out.String())
84 return err
85 }
86 err = run(true)
87 var status syscall.WaitStatus
88 if exit, ok := err.(*exec.ExitError); ok {
89 status = exit.Sys().(syscall.WaitStatus)
90 }
91 if !status.Signaled() || status.Signal() != syscall.SIGKILL {
92 t.Fatal("seed did not die at precommit", err)
93 }
94 if _, err := os.Stat(saved.Signal); err != nil {
95 t.Fatal("never reached materialization", err)
96 }
97 before, err := hostclaim.Read(t.Context(), claims, recoveryHost)
98 if err != nil || before.Claim.Holder.WakeRetry.State != "attempting" || before.Claim.Holder.WakeRetry.AttemptsReserved != 1 {
99 t.Fatal("crash lost retry debt", err)
100 }
101 data := f.f.data.(*leasetest.VersionedBucket)
102 first := movesFilesPrefix(before.Claim.Holder.Location, before.Claim.Holder.Generation)
103 if data.LatestID(f.actor.Bucket, hostfiles.ManifestKey(first)) != "" {
104 t.Fatal("crashed seed committed")
105 }
106 // Model expiry of the dead owner's remote lease; no production TTL is changed.
107 versions, err := data.ListVersions(t.Context(), f.actor.Bucket, before.Claim.Holder.Location+"/")
108 if err != nil {
109 t.Fatal(err)
110 }
111 for _, v := range versions {
112 if strings.HasSuffix(v.Key, "/lock.json") && v.IsLatest {
113 body, err := data.GetVersion(t.Context(), f.actor.Bucket, v.Key, v.VersionID)
114 if err != nil {
115 t.Fatal(err)
116 }
117 var lock map[string]any
118 if err := json.Unmarshal(body, &lock); err != nil {
119 t.Fatal(err)
120 }
121 lock["expires_at"] = time.Now().Add(-time.Minute).UTC().Format(time.RFC3339Nano)
122 expired, _ := json.Marshal(lock)
123 data.Put(f.actor.Bucket, v.Key, expired)
124 }
125 }
126 if err := run(false); err == nil {
127 t.Fatal("rerun bypassed settlement backoff")
128 }
129 settled, err := hostclaim.Read(t.Context(), claims, recoveryHost)
130 if err != nil || settled.Claim.Holder.WakeRetry.State != "open" || settled.Claim.Holder.WakeRetry.AttemptsReserved != 1 || settled.Claim.Holder.WakeRetry.Attempt != nil {
131 t.Fatal("rerun did not settle crashed attempt before backoff", err)
132 }
133 saved.Now = saved.Now.Add(time.Minute)
134 savedBody, _ = json.Marshal(saved)
135 if err := os.WriteFile(config, savedBody, 0600); err != nil {
136 t.Fatal(err)
137 }
138 if err := run(false); err != nil {
139 t.Fatal("rerun after backoff", err)
140 }
141 after, err := hostclaim.Read(t.Context(), claims, recoveryHost)
142 if err != nil || after.Claim.Holder.WakeRetry.State != "open" || after.Claim.Holder.WakeRetry.AttemptsReserved != 2 || after.Claim.Holder.Generation == before.Claim.Holder.Generation {
143 t.Fatal("rerun did not settle and reserve once", err)
144 }
145 final := movesFilesPrefix(after.Claim.Holder.Location, after.Claim.Holder.Generation)
146 if data.LatestID(f.actor.Bucket, hostfiles.ManifestKey(final)) == "" {
147 t.Fatal("rerun did not commit real admitted prefix")
148 }
149 etag := after.ETag
150 if err := run(false); err != nil {
151 t.Fatal(err)
152 }
153 after, err = hostclaim.Read(t.Context(), claims, recoveryHost)
154 if err != nil || after.ETag != etag {
155 t.Fatal("established rerun changed claim", err)
156 }
157}
158
159func TestRootFilesSeedWholeExecutorPayload(t *testing.T) {
160 path := os.Getenv("FILES_SEED_CRASH_CONFIG")
161 if path == "" {
162 t.Fatal("invoke through killed executor witness")
163 }
164 body, err := os.ReadFile(path)
165 if err != nil {
166 t.Fatal(err)
167 }
168 var saved seedCrashFixture
169 if err := json.Unmarshal(body, &saved); err != nil {
170 t.Fatal(err)
171 }
172 t.Setenv("AWS_REGION", "eu-west-1")
173 t.Setenv("AWS_DEFAULT_REGION", "eu-west-1")
174 claims := release.NewDirStore(saved.Claims)
175 data, err := lease.NewVersionedS3(t.Context(), map[string]string{saved.Actor.Bucket: "eu-west-1"})
176 if err != nil {
177 t.Fatal(err)
178 }
179 journal, err := hostbox.OpenLaunchJournal(saved.Journal)
180 if err != nil {
181 t.Fatal(err)
182 }
183 defer journal.Close()
184 roots := &servingAuthorityDisk{rollbackDisk: &rollbackDisk{dir: saved.Root}}
185 rf := &recoveryFixture{t: t, claims: claims, evidence: claims, journal: journal, journalPath: saved.Journal, data: data, bucket: saved.Actor.Bucket, signer: ed25519.NewKeyFromSeed(make([]byte, ed25519.SeedSize)), now: saved.Now, seq: os.Getpid() * 100000}
186 rf.guard = hostguard.Guard{DataDir: saved.Data, LockDir: saved.Locks, Claims: hostguard.StoreReader{Store: claims}, Root: hostguard.NewRootAuthority(roots)}
187 rf.kernel = &recoveryKernel{journal: journal}
188 crash := os.Getenv("FILES_SEED_CRASH") == "true"
189 if !crash {
190 rf.now = rf.now.Add(time.Minute)
191 }
192 reg, err := registry.Open(saved.Registry)
193 if err != nil {
194 t.Fatal(err)
195 }
196 defer reg.Close()
197 row, err := reg.Get(recoveryHost)
198 if err != nil || row == nil {
199 t.Fatal(err)
200 }
201 state, err := rootstate.Open(saved.Parks, false, false)
202 if err != nil {
203 t.Fatal(err)
204 }
205 defer state.Close()
206 journals, err := hostbox.NewJournalService(journal)
207 if err != nil {
208 t.Fatal(err)
209 }
210 f := &servingRootFixture{t: t, f: rf, rootDisk: roots, db: row.DBPath, actor: saved.Actor}
211 f.p = &servingParkRoot{ledger: &servingParkLedger{state: state, fs: newServingParkFS(saved.Data, state), journal: journal, now: func() time.Time { return rf.now }}, empty: plan9D5Fixture(t).owners}
212 coordinator := rf.root(nil)
213 coordinator.Kernel = &noProcessKernelFake{}
214 coordinator.Inventory = journals.Inventory
215 f.p.root = ClaimRootHTTP{Roots: map[string]string{"data": saved.Data}, Journals: journals, Launches: journal, Stop: ClaimStopHTTP{Coordinator: coordinator, Authenticate: func(context.Context, net.Conn) (StopPeer, error) { return rf.peer, nil }}, InventoryActor: func(context.Context) (hostbox.Actor, error) { return f.actor, nil }, RequireNewWork: func(context.Context) error { return nil }}
216 f.p.root.Reconcile.Registry = reg
217 f.p.root.Reconcile.Inventory = journals.Inventory
218 f.p.root.Reconcile.Box, f.p.root.Reconcile.Enrollment = f.actor.Box, f.actor.Enrollment
219 f.p.root.Reconcile.Retirement.JournalDir = saved.Retirement
220 p := newServingPreparation(t, f, "boot", fmt.Sprintf("seed-%d", os.Getpid()))
221 a := p.assembly
222 a.Data = data
223 a.Identity = handoverRecoveryIdentity(t, saved.Actor)
224 source := &movesFilesSource{DataDir: saved.Data, Routes: reg, newFiles: func(ctx context.Context, _, _ string) (release.Store, error) {
225 return release.NewS3Store(ctx, saved.Actor.Bucket)
226 }}
227 a.Wake.seedFactory = func(ctx context.Context, op *filesSeedOperation, token *activator.SeedExclusion) (*activator.ClaimRuntime, *ClaimWakeAssembly, error) {
228 runtime, wake, err := newFilesSeedRuntime(ctx, a, p.source, p.writer, op, token)
229 if err != nil {
230 return runtime, wake, err
231 }
232 prepare := runtime.Prepare
233 runtime.Prepare = func(ctx context.Context, rt registry.Route, o hostclaim.Observed, policy activator.StartPolicy) (activator.ClaimStart, error) {
234 start, err := prepare(ctx, rt, o, policy)
235 if err == nil && crash {
236 start.Materialize = func(context.Context) error {
237 if err := os.WriteFile(saved.Signal, []byte(o.Claim.Holder.Generation), 0600); err != nil {
238 return err
239 }
240 syscall.Kill(os.Getpid(), syscall.SIGKILL)
241 select {}
242 }
243 }
244 return start, err
245 }
246 return runtime, wake, nil
247 }
248 executor := &filesSeedExecutor{assembly: &a, service: &activator.Service{}, registry: reg, source: source}
249 if err := executor.run(t.Context(), recoveryHost, os.Stdout); err != nil {
250 t.Fatal(err)
251 }
252}
253