amadan

carlos / platform Public

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

Plain git — no account needed to clone.

Download

Download this file

1package lease
2
3// CARLOS_S3_ENDPOINT tests. Unlike lease_test.go these need no AWS and no
4// CARLOS_TEST_S3_BUCKET: the point of the override is that S3 traffic can be
5// pointed anywhere, so the test points it at an httptest server and asserts
6// the request actually arrives. A regression here — the leaser silently going
7// back to real AWS — is exactly what cost this deployment hibernation and the
8// whole `carlos instances create` provisioning path.
9
10import (
11 "context"
12 "encoding/json"
13 "errors"
14 "fmt"
15 "io"
16 "net/http"
17 "net/http/httptest"
18 "os"
19 "path/filepath"
20 "strings"
21 "sync"
22 "testing"
23
24 "github.com/aws/aws-sdk-go-v2/aws"
25 "github.com/benbjohnson/litestream"
26 lss3 "github.com/benbjohnson/litestream/s3"
27 "github.com/carlosframework/platform/internal/rootaws"
28)
29
30// fakeS3 is just enough of S3 for the leaser's three verbs (GET/PUT/DELETE on
31// one key) to complete a full acquire-then-release, recording every request so
32// a test can assert on where the traffic went.
33type fakeS3 struct {
34 mu sync.Mutex
35 reqs []string // "METHOD /path"
36 body []byte // last PUT body
37}
38
39func (f *fakeS3) record(method, path string) {
40 f.mu.Lock()
41 defer f.mu.Unlock()
42 f.reqs = append(f.reqs, method+" "+path)
43}
44
45func (f *fakeS3) requests() []string {
46 f.mu.Lock()
47 defer f.mu.Unlock()
48 return append([]string(nil), f.reqs...)
49}
50
51func (f *fakeS3) putBody() []byte {
52 f.mu.Lock()
53 defer f.mu.Unlock()
54 return append([]byte(nil), f.body...)
55}
56
57func newFakeS3(t *testing.T) (*fakeS3, string) {
58 t.Helper()
59 f := &fakeS3{}
60 srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
61 f.record(r.Method, r.URL.Path)
62 switch r.Method {
63 case http.MethodGet:
64 // No lease on the key yet. The XML body matters: litestream
65 // classifies "absent" by the S3 error CODE, and a bare 404 with no
66 // body would read as a hard failure instead.
67 w.Header().Set("Content-Type", "application/xml")
68 w.WriteHeader(http.StatusNotFound)
69 _, _ = w.Write([]byte(`<?xml version="1.0" encoding="UTF-8"?>` +
70 `<Error><Code>NoSuchKey</Code><Message>no such key</Message></Error>`))
71 case http.MethodPut:
72 b, _ := io.ReadAll(r.Body)
73 f.mu.Lock()
74 f.body = b
75 f.mu.Unlock()
76 w.Header().Set("ETag", `"lease-etag"`)
77 w.WriteHeader(http.StatusOK)
78 case http.MethodDelete:
79 w.WriteHeader(http.StatusNoContent)
80 default:
81 w.WriteHeader(http.StatusMethodNotAllowed)
82 }
83 }))
84 t.Cleanup(srv.Close)
85 return f, srv.URL
86}
87
88// fakeCreds plants credentials the AWS default chain will resolve without
89// touching the operator's real profile or IMDS — the same isolation instinct
90// as carlosTestEnv, applied to the SDK instead of the carlos binary.
91func fakeCreds(t *testing.T) {
92 t.Helper()
93 t.Setenv("AWS_ACCESS_KEY_ID", "test-key")
94 t.Setenv("AWS_SECRET_ACCESS_KEY", "test-secret")
95 t.Setenv("AWS_SESSION_TOKEN", "")
96 // Cleared, not left ambient: region resolution is under test, so a region
97 // inherited from whatever machine runs the suite would make these results
98 // depend on the box. Tests that care set it back explicitly.
99 t.Setenv("AWS_REGION", "")
100 t.Setenv("AWS_DEFAULT_REGION", "")
101 t.Setenv("AWS_EC2_METADATA_DISABLED", "true")
102}
103
104// TestAcquire_ReachesEndpointOverride is the regression test for the bug this
105// file exists for: with CARLOS_S3_ENDPOINT set, the lease's lock.json must be
106// written through the override — not to real AWS. It asserts on the recorded
107// requests, so it fails loudly (rather than hanging on a live AWS dial) if the
108// leaser ever stops honouring the endpoint.
109func TestAcquire_ReachesEndpointOverride(t *testing.T) {
110 f, url := newFakeS3(t)
111 fakeCreds(t)
112 t.Setenv("CARLOS_S3_ENDPOINT", url)
113
114 // Region deliberately empty, and fakeCreds leaves the chain empty too, so
115 // this also exercises the last resort: a client signing for no region at
116 // all would fail every request against the fake.
117 held, err := Acquire(context.Background(), "test-bucket", "inst/example.com", "")
118 if err != nil {
119 t.Fatalf("Acquire against %s: %v", url, err)
120 }
121
122 got := f.requests()
123 // Path-style addressing: bucket in the path, not the hostname — a
124 // compatible store has no per-bucket virtual host.
125 wantKey := "/test-bucket/inst/example.com/lock.json"
126 var sawGet, sawPut bool
127 for _, r := range got {
128 switch r {
129 case "GET " + wantKey:
130 sawGet = true
131 case "PUT " + wantKey:
132 sawPut = true
133 }
134 }
135 // An absent lock is one create-only PUT (createLease); a GET comes
136 // only when a lock is present. Either way the PUT must arrive here.
137 _ = sawGet
138 if !sawPut {
139 t.Fatalf("lease traffic did not reach the override: got %v, want a PUT of %s", got, wantKey)
140 }
141
142 // The PUT body is the lock itself — proof the acquire really happened
143 // here rather than the server merely being probed.
144 var lease litestream.Lease
145 if err := json.Unmarshal(f.putBody(), &lease); err != nil {
146 t.Fatalf("decode written lock.json (%q): %v", f.putBody(), err)
147 }
148 if lease.Owner == "" {
149 t.Errorf("written lease carries no owner: %+v", lease)
150 }
151
152 if err := held.Release(context.Background()); err != nil {
153 t.Fatalf("Release: %v", err)
154 }
155 var sawDelete bool
156 for _, r := range f.requests() {
157 if r == "DELETE "+wantKey {
158 sawDelete = true
159 }
160 }
161 if !sawDelete {
162 t.Errorf("release did not reach the override: got %v", f.requests())
163 }
164}
165
166// TestNewLeaserClient_EndpointConfig pins the client options either side of the
167// override, because the no-override case is the flagship's production path and
168// must stay byte-identical to what it was before endpoint.go existed.
169func TestNewLeaserClient_EndpointConfig(t *testing.T) {
170 fakeCreds(t)
171
172 t.Run("override sets a path-style base endpoint", func(t *testing.T) {
173 t.Setenv("CARLOS_S3_ENDPOINT", "https://fsn1.your-objectstorage.com")
174 c, _, err := newLeaserClient(context.Background(), "eu-central-1")
175 if err != nil {
176 t.Fatalf("newLeaserClient: %v", err)
177 }
178 o := c.Options()
179 if o.BaseEndpoint == nil || *o.BaseEndpoint != "https://fsn1.your-objectstorage.com" {
180 t.Errorf("BaseEndpoint = %v, want the override", o.BaseEndpoint)
181 }
182 if !o.UsePathStyle {
183 t.Error("UsePathStyle = false, want true under an endpoint override")
184 }
185 })
186
187 t.Run("no override leaves stock AWS addressing", func(t *testing.T) {
188 t.Setenv("CARLOS_S3_ENDPOINT", "")
189 c, _, err := newLeaserClient(context.Background(), "eu-west-1")
190 if err != nil {
191 t.Fatalf("newLeaserClient: %v", err)
192 }
193 o := c.Options()
194 if o.BaseEndpoint != nil {
195 t.Errorf("BaseEndpoint = %q, want nil (stock AWS)", *o.BaseEndpoint)
196 }
197 if o.UsePathStyle {
198 t.Error("UsePathStyle = true, want false without an override")
199 }
200 })
201}
202
203// TestNewReplicaClient_EndpointConfig covers the other half of a lease: the
204// LTX files beside lock.json. A replica client that ignored the override while
205// the leaser honoured it would split one lease across two stores.
206func TestNewReplicaClient_EndpointConfig(t *testing.T) {
207 fakeCreds(t)
208 t.Run("override", func(t *testing.T) {
209 t.Setenv("CARLOS_S3_ENDPOINT", "https://fsn1.your-objectstorage.com")
210 c, err := newReplicaClient("b", "p", "eu-central-1")
211 if err != nil {
212 t.Fatal(err)
213 }
214 if c.Endpoint != "https://fsn1.your-objectstorage.com" {
215 t.Errorf("Endpoint = %q, want the override", c.Endpoint)
216 }
217 if !c.ForcePathStyle {
218 t.Error("ForcePathStyle = false, want true under an endpoint override")
219 }
220 if c.Bucket != "b" || c.Path != "p" || c.Region != "eu-central-1" {
221 t.Errorf("bucket/path/region = %q/%q/%q, want b/p/eu-central-1", c.Bucket, c.Path, c.Region)
222 }
223 })
224
225 t.Run("no override", func(t *testing.T) {
226 t.Setenv("CARLOS_S3_ENDPOINT", "")
227 c, err := newReplicaClient("b", "p", "eu-west-1")
228 if err != nil {
229 t.Fatal(err)
230 }
231 if c.Endpoint != "" || c.ForcePathStyle {
232 t.Errorf("Endpoint/ForcePathStyle = %q/%v, want empty/false without an override", c.Endpoint, c.ForcePathStyle)
233 }
234 })
235}
236
237// TestRegionFor pins the order of authority. The case that matters most is
238// "chain wins over the non-AWS default": a compatible-store deployment sets
239// AWS_REGION to its own location (carlos.exciting.io uses nbg1) and that is
240// what release.NewS3Store already signs for against the same bucket —
241// substituting a constant would make the lease sign differently from every
242// other S3 client in the process.
243func TestRegionFor(t *testing.T) {
244 tests := []struct {
245 name, endpoint, explicit, chain, want string
246 }{
247 {"explicit wins over everything", "https://s3.example.com", "fsn1", "nbg1", "fsn1"},
248 {"chain wins over the non-AWS default", "https://s3.example.com", "", "nbg1", "nbg1"},
249 {"non-AWS default only fills a total blank", "https://s3.example.com", "", "", lss3.DefaultRegion},
250 {"no override leaves a blank blank", "", "", "", ""},
251 {"no override keeps the chain", "", "", "eu-west-1", "eu-west-1"},
252 }
253 for _, tt := range tests {
254 t.Run(tt.name, func(t *testing.T) {
255 t.Setenv("CARLOS_S3_ENDPOINT", tt.endpoint)
256 if got := regionFor(tt.explicit, tt.chain); got != tt.want {
257 t.Errorf("regionFor(%q, %q) with endpoint %q = %q, want %q",
258 tt.explicit, tt.chain, tt.endpoint, got, tt.want)
259 }
260 })
261 }
262}
263
264// TestResolveRegion_PrefersAWSRegionEnv is the end-to-end version of the case
265// above, through the real AWS chain: with AWS_REGION set and no explicit
266// region, the lease must sign for AWS_REGION — not for litestream's non-AWS
267// constant. This is the exact shape of the private CARLOS box's host.env.
268func TestResolveRegion_PrefersAWSRegionEnv(t *testing.T) {
269 fakeCreds(t)
270 t.Setenv("CARLOS_S3_ENDPOINT", "https://nbg1.your-objectstorage.com")
271 t.Setenv("AWS_REGION", "nbg1")
272
273 got, err := resolveRegion(context.Background(), "")
274 if err != nil {
275 t.Fatalf("resolveRegion: %v", err)
276 }
277 if got != "nbg1" {
278 t.Errorf("resolveRegion = %q, want %q (AWS_REGION must outrank the non-AWS default)", got, "nbg1")
279 }
280}
281
282// TestRestoreCreatesTheOutputPathsParentDirectory pins the review fix: a
283// nested row's instance directory does not exist before its first wake (it
284// is created by the spawn path, which runs after Restore), so Restore must
285// create OutputPath's parent itself rather than fail with ENOENT before the
286// spawn path ever gets a chance to.
287func TestRestoreCreatesTheOutputPathsParentDirectory(t *testing.T) {
288 fakeCreds(t)
289 // Every listing this exercises, the v0.3.x/LTX format comparison and
290 // the LTX restore plan itself, is a ListObjectsV2 call; answering all
291 // of them with an empty bucket makes Restore take the "no replica"
292 // path, the one every fresh nested row actually takes on its first wake.
293 srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
294 w.Header().Set("Content-Type", "application/xml")
295 fmt.Fprint(w, `<?xml version="1.0" encoding="UTF-8"?>
296<ListBucketResult xmlns="http://s3.amazonaws.com/doc/2006-03-01/"><Name>b</Name><KeyCount>0</KeyCount><IsTruncated>false</IsTruncated></ListBucketResult>`)
297 }))
298 defer srv.Close()
299 t.Setenv("CARLOS_S3_ENDPOINT", srv.URL)
300
301 root := t.TempDir()
302 dbPath := filepath.Join(root, "instances", "h.example.com", "h.example.com.db")
303 if _, err := os.Stat(filepath.Dir(dbPath)); !os.IsNotExist(err) {
304 t.Fatalf("test setup: parent directory already exists")
305 }
306
307 restored, err := Restore(context.Background(), dbPath, "test-bucket", "inst/h.example.com", "")
308 if err != nil {
309 t.Fatalf("Restore into a path whose parent is absent: %v", err)
310 }
311 if restored {
312 t.Fatal("restored = true against an empty bucket, want false (nothing to restore)")
313 }
314 if fi, err := os.Stat(filepath.Dir(dbPath)); err != nil || !fi.IsDir() {
315 t.Fatalf("Restore did not create the output path's parent directory: %v", err)
316 }
317}
318
319// ReadReplicaClient is the db door's reader (spec 2026-09-02 §6). It must
320// carry the coordinates it was given and sign for the region lease.Restore
321// would: an explicit region wins, and an empty one falls to the same chain
322// resolution the writer uses — the endpoint override's non-AWS default when
323// nothing else answers.
324func TestReadReplicaClientCarriesBucketPrefixRegion(t *testing.T) {
325 fakeCreds(t)
326 c, err := ReadReplicaClient(context.Background(), "data-bucket", "instances/a.example.com", "eu-west-1")
327 if err != nil {
328 t.Fatal(err)
329 }
330 if c.Bucket != "data-bucket" {
331 t.Errorf("Bucket = %q, want data-bucket", c.Bucket)
332 }
333 if c.Path != "instances/a.example.com" {
334 t.Errorf("Path = %q, want instances/a.example.com", c.Path)
335 }
336 if c.Region != "eu-west-1" {
337 t.Errorf("Region = %q, want eu-west-1", c.Region)
338 }
339}
340
341func TestReadReplicaClientResolvesAnEmptyRegionLikeRestore(t *testing.T) {
342 fakeCreds(t)
343 t.Setenv("CARLOS_S3_ENDPOINT", "http://127.0.0.1:1")
344 c, err := ReadReplicaClient(context.Background(), "data-bucket", "instances/a.example.com", "")
345 if err != nil {
346 t.Fatal(err)
347 }
348 if c.Region != lss3.DefaultRegion {
349 t.Errorf("Region = %q, want litestream's non-AWS default %q (the writer's resolution)", c.Region, lss3.DefaultRegion)
350 }
351 if c.Endpoint != "http://127.0.0.1:1" || !c.ForcePathStyle {
352 t.Errorf("Endpoint = %q ForcePathStyle = %v, want the override applied", c.Endpoint, c.ForcePathStyle)
353 }
354}
355
356func TestRootReplicaClientEnvironmentKeepsDefaultChain(t *testing.T) {
357 for _, region := range []string{"eu-west-1", ""} {
358 t.Run("region="+region, func(t *testing.T) {
359 fakeCreds(t)
360 t.Setenv("AWS_SESSION_TOKEN", "test-session-token")
361 t.Setenv("AWS_PROFILE", "")
362 t.Setenv("AWS_DEFAULT_PROFILE", "")
363 dir := t.TempDir()
364 marker := filepath.Join(dir, "executed")
365 config := filepath.Join(dir, "config")
366 if err := os.WriteFile(config, []byte("[default]\ncredential_process = touch "+marker+"\n"), 0600); err != nil {
367 t.Fatal(err)
368 }
369 t.Setenv("AWS_CONFIG_FILE", config)
370 t.Setenv("AWS_SHARED_CREDENTIALS_FILE", config)
371 var mu sync.Mutex
372 var locations, listings int
373 srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
374 if !strings.Contains(r.Header.Get("Authorization"), "Credential=test-key/") || r.Header.Get("X-Amz-Security-Token") != "test-session-token" {
375 t.Error("request did not carry the explicit keys and session token")
376 }
377 w.Header().Set("Content-Type", "application/xml")
378 mu.Lock()
379 defer mu.Unlock()
380 if r.URL.Query().Has("location") {
381 locations++
382 fmt.Fprint(w, `<LocationConstraint xmlns="http://s3.amazonaws.com/doc/2006-03-01/">eu-west-1</LocationConstraint>`)
383 } else {
384 listings++
385 fmt.Fprint(w, `<ListBucketResult xmlns="http://s3.amazonaws.com/doc/2006-03-01/"><Name>b</Name><KeyCount>0</KeyCount><IsTruncated>false</IsTruncated></ListBucketResult>`)
386 }
387 }))
388 defer srv.Close()
389 // Keep Endpoint empty to exercise Litestream's bucket-region lookup.
390 t.Setenv("CARLOS_S3_ENDPOINT", "")
391 t.Setenv("AWS_ENDPOINT_URL_S3", srv.URL)
392 c, err := newReplicaClientForUID("b", "p", region, 0, rootaws.OffcloudCredentials)
393 if err != nil {
394 t.Fatal(err)
395 }
396 c.ForcePathStyle = true
397 if c.AccessKeyID != "" || c.SecretAccessKey != "" {
398 t.Fatal("environment credentials must use the upstream default chain")
399 }
400 if _, err := c.GenerationsV3(context.Background()); err != nil {
401 t.Fatal(err)
402 }
403 mu.Lock()
404 defer mu.Unlock()
405 if listings == 0 || (region == "" && locations != 1) || (region != "" && locations != 0) {
406 t.Fatalf("locations=%d listings=%d for region %q", locations, listings, region)
407 }
408 if _, err := os.Stat(marker); !os.IsNotExist(err) {
409 t.Fatal("credential_process executed")
410 }
411 })
412 }
413}
414
415func TestNonRootReplicaClientKeepsDefaultProvider(t *testing.T) {
416 c, err := newReplicaClientForUID("b", "p", "eu-west-1", 1001, func() (aws.Credentials, error) {
417 t.Fatal("non-root read root credentials")
418 return aws.Credentials{}, nil
419 })
420 if err != nil {
421 t.Fatal(err)
422 }
423 if c.AccessKeyID != "" || c.SecretAccessKey != "" {
424 t.Fatal("non-root credential selection changed")
425 }
426}
427
428func TestRootReplicaClientOffcloudCredentials(t *testing.T) {
429 for _, token := range []string{"", "FILE_TOKEN"} {
430 t.Run("token="+token, func(t *testing.T) {
431 body := "[default]\naws_access_key_id=FILE_ACCESS\naws_secret_access_key=FILE_SECRET\n"
432 if token != "" {
433 body += "aws_session_token=" + token + "\n"
434 }
435 c, err := newReplicaClientForUID("b", "p", "eu-west-1", 0, func() (aws.Credentials, error) {
436 return rootaws.ParseStaticCredentials([]byte(body))
437 })
438 if token != "" {
439 if !errors.Is(err, errRootReplicaSessionToken) || c != nil {
440 t.Fatalf("token-bearing file: client=%v err=%v", c, err)
441 }
442 return
443 }
444 if err != nil {
445 t.Fatal(err)
446 }
447 if c.AccessKeyID != "FILE_ACCESS" || c.SecretAccessKey != "FILE_SECRET" {
448 t.Fatal("off-cloud file keys were not set on the upstream client")
449 }
450 })
451 }
452}
453
454func TestRootReplicaClientWithoutOffcloudKeepsDefaultChain(t *testing.T) {
455 c, err := newReplicaClientForUID("b", "p", "eu-west-1", 0, func() (aws.Credentials, error) {
456 return aws.Credentials{}, nil
457 })
458 if err != nil {
459 t.Fatal(err)
460 }
461 if c.AccessKeyID != "" || c.SecretAccessKey != "" {
462 t.Fatal("instance-profile credentials must use the upstream default chain")
463 }
464}
465
466func TestRootReplicaClientRefusesUnsafeOffcloudCredentials(t *testing.T) {
467 refused := errors.New("unsafe file")
468 c, err := newReplicaClientForUID("b", "p", "eu-west-1", 0, func() (aws.Credentials, error) {
469 return aws.Credentials{}, refused
470 })
471 if !errors.Is(err, refused) || c != nil {
472 t.Fatalf("unsafe file: client=%v err=%v", c, err)
473 }
474}
475