diff --git a/sandboxd/pool/promote_test.go b/sandboxd/pool/promote_test.go index 0da536f93..9bd4f1c2b 100644 --- a/sandboxd/pool/promote_test.go +++ b/sandboxd/pool/promote_test.go @@ -2,6 +2,7 @@ package pool import ( "bytes" + "context" "errors" "maps" "os" @@ -12,6 +13,7 @@ import ( "github.com/cocoonstack/sandbox/sandboxd/config" "github.com/cocoonstack/sandbox/sandboxd/store" + "github.com/cocoonstack/sandbox/sandboxd/store/storetest" "github.com/cocoonstack/sandbox/sandboxd/tenants" "github.com/cocoonstack/sandbox/sandboxd/types" ) @@ -478,7 +480,7 @@ func TestTemplatesListWhatTheNodeHoldsAcrossARestart(t *testing.T) { if err = os.WriteFile(filepath.Join(staging, store.MetaFile), []byte(`{"id":"`+oldID+`","created_at":"2026-01-01T00:00:00Z"}`), 0o600); err != nil { t.Fatalf("write keyless meta: %v", err) } - if _, err = m.tpls.PublishDigested(t.Context(), staging, oldID, ""); err != nil { + if _, err = storetest.PublishDigested(t.Context(), m.tpls, staging, oldID); err != nil { t.Fatalf("publish keyless record: %v", err) } @@ -594,6 +596,43 @@ func TestRejectedLabelWritesLeakNoRecLock(t *testing.T) { } } +func TestClaimFromATemplateRunsBesideItsRepromote(t *testing.T) { + m := newTestManager(t, newFakeEngine()) + parent := mustClaim(t, m, testKey) + key, _, err := m.Promote(t.Context(), parent.ID, Cred{Token: parent.Token}, "tpl:x", "") + if err != nil { + t.Fatalf("Promote: %v", err) + } + slow := &slowPrepareStore{Store: m.tpls, entered: make(chan struct{}), release: make(chan struct{})} + m.tpls = slow + repromoted := make(chan error, 1) + go func() { + _, _, promoteErr := m.Promote(t.Context(), parent.ID, Cred{Token: parent.Token}, "tpl:x", "") + repromoted <- promoteErr + }() + <-slow.entered + + ctx, cancel := context.WithTimeout(t.Context(), 100*time.Millisecond) + defer cancel() + claimed := make(chan error, 1) + go func() { + _, claimErr := claimAny(ctx, m, key, 0) + claimed <- claimErr + }() + select { + case err = <-claimed: + if err != nil { + t.Errorf("claim during a re-promote: %v, want a clone of the committed template", err) + } + case <-time.After(2 * time.Second): + t.Error("a claim with a 100 ms context was still blocked 2 s into a re-promote") + } + close(slow.release) + if err = <-repromoted; err != nil { + t.Fatalf("re-promote: %v", err) + } +} + func claimTenant(t *testing.T, m *Manager, tenant string) *types.Sandbox { t.Helper() sb, err := m.ClaimProvision(t.Context(), testKey, ClaimOptions{TTL: time.Hour, Tenant: tenant}) @@ -614,3 +653,15 @@ func stageTemplate(t *testing.T, m *Manager, id string) string { } return staging } + +type slowPrepareStore struct { + store.Store + entered chan struct{} + release chan struct{} +} + +func (s *slowPrepareStore) PrepareDigested(ctx context.Context, staging, id, lineage string) (string, store.CommitFunc, error) { + close(s.entered) + <-s.release + return s.Store.PrepareDigested(ctx, staging, id, lineage) +} diff --git a/sandboxd/pool/template.go b/sandboxd/pool/template.go index e3307e2df..f6b7e5750 100644 --- a/sandboxd/pool/template.go +++ b/sandboxd/pool/template.go @@ -300,10 +300,6 @@ func (m *Manager) checkTemplateOwner(ctx context.Context, id, tenant string) err func (m *Manager) commitTemplate(ctx context.Context, staging string, rec templateRecord) (string, error) { id := store.TemplateID(rec.Key.Hash()) - defer m.lockRec(id)() - if ownerErr := m.checkTemplateOwner(ctx, id, rec.Tenant); ownerErr != nil { - return "", ownerErr - } rec.ID, rec.CreatedAt = id, time.Now() meta, err := json.Marshal(rec) if err != nil { @@ -312,10 +308,17 @@ func (m *Manager) commitTemplate(ctx context.Context, staging string, rec templa if err = os.WriteFile(filepath.Join(staging, store.MetaFile), meta, 0o600); err != nil { return "", err } - digest, err := m.tpls.PublishDigested(ctx, staging, id, "") + digest, commit, err := m.tpls.PrepareDigested(ctx, staging, id, "") if err != nil { return "", fmt.Errorf("publish template: %w", err) } + defer m.lockRec(id)() + if ownerErr := m.checkTemplateOwner(ctx, id, rec.Tenant); ownerErr != nil { + return "", ownerErr + } + if err = commit(ctx); err != nil { + return "", fmt.Errorf("publish template: %w", err) + } if err = m.tpls.SetLabels(ctx, id, nil); err != nil { log.WithFunc("pool.commitTemplate").Warnf(ctx, "clear the previous promote's labels on %s: %v", id, err) } diff --git a/sandboxd/store/dir/dir.go b/sandboxd/store/dir/dir.go index fc70fbf85..ec4f1f82a 100644 --- a/sandboxd/store/dir/dir.go +++ b/sandboxd/store/dir/dir.go @@ -60,15 +60,12 @@ func (d *Store) Publish(ctx context.Context, staging, id, _ string) error { return d.publish(ctx, staging, id, "") } -func (d *Store) PublishDigested(ctx context.Context, staging, id, _ string) (string, error) { +func (d *Store) PrepareDigested(ctx context.Context, staging, id, _ string) (string, store.CommitFunc, error) { digest, err := hashExport(ctx, filepath.Join(staging, store.ExportDir)) if err != nil { - return "", fmt.Errorf("digest export: %w", err) + return "", nil, fmt.Errorf("digest export: %w", err) } - if err := d.publish(ctx, staging, id, digest); err != nil { - return "", fmt.Errorf("publish digested record: %w", err) - } - return digest, nil + return digest, func(ctx context.Context) error { return d.publish(ctx, staging, id, digest) }, nil } func (d *Store) Fetch(ctx context.Context, id string) (string, []byte, string, error) { diff --git a/sandboxd/store/dir/dir_test.go b/sandboxd/store/dir/dir_test.go index 144ed61cf..e8e68f4ca 100644 --- a/sandboxd/store/dir/dir_test.go +++ b/sandboxd/store/dir/dir_test.go @@ -134,7 +134,7 @@ func TestPublishDigestedChunkBoundaries(t *testing.T) { if err = os.WriteFile(filepath.Join(staging, store.MetaFile), []byte(metaJSON(tt.name)), 0o600); err != nil { t.Fatalf("write meta: %v", err) } - digest, err := st.PublishDigested(t.Context(), staging, id, "") + digest, err := storetest.PublishDigested(t.Context(), st, staging, id) if err != nil { t.Fatalf("PublishDigested: %v", err) } @@ -216,7 +216,7 @@ func TestPublishDigestedFailureAndFreshRetry(t *testing.T) { if err = os.Mkdir(blocker, 0o750); err != nil { t.Fatalf("block digest commit: %v", err) } - if digest, publishErr := st.PublishDigested(t.Context(), failed, id, ""); publishErr == nil || digest != "" { + if digest, publishErr := storetest.PublishDigested(t.Context(), st, failed, id); publishErr == nil || digest != "" { t.Fatalf("PublishDigested with blocked sidecar = %q, %v, want empty/error", digest, publishErr) } dir, meta, digest, err := st.Fetch(t.Context(), id) @@ -658,7 +658,7 @@ func mustPublishDigested(t *testing.T, st *Store, id, metaID string) string { t.Fatalf("stage %s: %v", metaID, err) } seedRecord(t, staging, metaID) - digest, err := st.PublishDigested(t.Context(), staging, id, "") + digest, err := storetest.PublishDigested(t.Context(), st, staging, id) if err != nil { t.Fatalf("publish digested %s: %v", metaID, err) } diff --git a/sandboxd/store/s3/s3.go b/sandboxd/store/s3/s3.go index 58b7e0fc6..3babe69f1 100644 --- a/sandboxd/store/s3/s3.go +++ b/sandboxd/store/s3/s3.go @@ -132,12 +132,15 @@ func (s *Store) Stage(id string) (string, error) { } func (s *Store) Publish(ctx context.Context, staging, id, lineage string) error { - _, err := s.publish(ctx, staging, id, lineage, false) - return err + _, commit, err := s.prepare(ctx, staging, id, lineage, false) + if err != nil { + return err + } + return commit(ctx) } -func (s *Store) PublishDigested(ctx context.Context, staging, id, lineage string) (string, error) { - return s.publish(ctx, staging, id, lineage, true) +func (s *Store) PrepareDigested(ctx context.Context, staging, id, lineage string) (string, store.CommitFunc, error) { + return s.prepare(ctx, staging, id, lineage, true) } func (s *Store) Fetch(ctx context.Context, id string) (string, []byte, string, error) { @@ -325,22 +328,22 @@ func (s *Store) get(ctx context.Context, key string) (*awss3.GetObjectOutput, er return out, nil } -func (s *Store) publish(ctx context.Context, staging, id, lineage string, digested bool) (string, error) { +func (s *Store) prepare(ctx context.Context, staging, id, lineage string, digested bool) (string, store.CommitFunc, error) { ctx, cancel := context.WithTimeout(ctx, transferBudget) defer cancel() lineage = cmp.Or(lineage, id) metaRaw, err := os.ReadFile(filepath.Join(staging, store.MetaFile)) //nolint:gosec // our own staging dir if err != nil { - return "", fmt.Errorf("staging has no %s: %w", store.MetaFile, err) + return "", nil, fmt.Errorf("staging has no %s: %w", store.MetaFile, err) } gen := store.ExportGen(metaRaw) files, err := publishFiles(staging) if err != nil { - return "", err + return "", nil, err } have, err := s.referencedChunks(ctx, id, lineage) if err != nil { - return "", err + return "", nil, err } man := manifest{Lineage: lineage, Files: make([]manifestFile, len(files))} digestFiles := make([]store.DigestFile, len(files)) @@ -360,18 +363,18 @@ func (s *Store) publish(ctx context.Context, staging, id, lineage string, digest err = fileErr } if err != nil { - return "", err + return "", nil, err } manifestRaw, err := json.Marshal(man) if err != nil { - return "", fmt.Errorf("encode manifest: %w", err) + return "", nil, fmt.Errorf("encode manifest: %w", err) } if err = s.putObject(ctx, s.manifestKey(id, gen), manifestRaw, nil); err != nil { - return "", err + return "", nil, err } if lineage != id { if err = s.putObject(ctx, s.lineageKey(lineage, id), nil, nil); err != nil { - return "", err + return "", nil, err } } digest := "" @@ -379,18 +382,19 @@ func (s *Store) publish(ctx context.Context, staging, id, lineage string, digest if digested { digest, err = store.AssembleDigest(digestFiles) if err != nil { - return "", err + return "", nil, err } metadata = map[string]string{digestMetadataKey: digest} } - if err = s.putObject(ctx, s.key(id, store.MetaFile), metaRaw, metadata); err != nil { - return "", err - } - // Keep old generations: another node may have selected the previous meta. - if err := os.RemoveAll(staging); err != nil { - return "", err - } - return digest, nil + return digest, func(ctx context.Context) error { + ctx, cancel := context.WithTimeout(ctx, transferBudget) + defer cancel() + if err := s.putObject(ctx, s.key(id, store.MetaFile), metaRaw, metadata); err != nil { + return err + } + // Keep old generations: another node may have selected the previous meta. + return os.RemoveAll(staging) + }, nil } // populate downloads one cache generation and installs it atomically. diff --git a/sandboxd/store/s3/s3_test.go b/sandboxd/store/s3/s3_test.go index 661cf605a..6a0e7b1ef 100644 --- a/sandboxd/store/s3/s3_test.go +++ b/sandboxd/store/s3/s3_test.go @@ -192,7 +192,7 @@ func TestPublishDigestedRetryAndMetaRequestAccounting(t *testing.T) { fake.setPutFailures(exportKey, 1) staging := stageRecord(t, st, id, meta, content) - digest, err := st.PublishDigested(t.Context(), staging, id, "") + digest, err := storetest.PublishDigested(t.Context(), st, staging, id) if err != nil { t.Fatalf("PublishDigested: %v", err) } @@ -254,7 +254,7 @@ func TestPublishDigestedFailurePreservesCommittedGeneration(t *testing.T) { st := newTestStore(t, fake) oldMeta := []byte(`{"id":"` + id + `","gen":1}`) oldStaging := stageRecord(t, st, id, oldMeta, "old") - oldDigest, err := st.PublishDigested(t.Context(), oldStaging, id, "") + oldDigest, err := storetest.PublishDigested(t.Context(), st, oldStaging, id) if err != nil { t.Fatalf("publish old generation: %v", err) } @@ -266,7 +266,7 @@ func TestPublishDigestedFailurePreservesCommittedGeneration(t *testing.T) { metaKey := "ck/" + id + "/" + store.MetaFile markerPuts := fake.requestCount(http.MethodPut, metaKey) fake.setPutFailures(newExportKey, 100) - digest, err := st.PublishDigested(t.Context(), newStaging, id, "") + digest, err := storetest.PublishDigested(t.Context(), st, newStaging, id) if err == nil { t.Fatal("PublishDigested succeeded with injected export failure") } @@ -290,7 +290,7 @@ func TestPublishDigestedFailurePreservesCommittedGeneration(t *testing.T) { } fake.setPutFailures(newExportKey, 0) - newDigest, err := st.PublishDigested(t.Context(), newStaging, id, "") + newDigest, err := storetest.PublishDigested(t.Context(), st, newStaging, id) if err != nil { t.Fatalf("retry PublishDigested: %v", err) } @@ -534,7 +534,7 @@ func TestPublishDigestedHashesHolesLikeZeros(t *testing.T) { size := 3*store.DigestChunkSize + 7 staging, src := stageSparseRecord(t, st, id, []byte(`{"id":"`+id+`"}`), size, map[int64]string{store.DigestChunkSize + 5: "data", size - 1: "z"}) sums := denseChunkSums(t, src) - digest, err := st.PublishDigested(t.Context(), staging, id, "") + digest, err := storetest.PublishDigested(t.Context(), st, staging, id) if err != nil { t.Fatalf("PublishDigested: %v", err) } @@ -552,7 +552,7 @@ func TestPublishDigestedSplitsChunksAtTheFixedBoundary(t *testing.T) { st := newTestStore(t, &fakeS3{objects: map[string][]byte{}}) content := string(bytes.Repeat([]byte{'x'}, int(store.DigestChunkSize))) + "y" staging := stageRecord(t, st, id, []byte(`{"id":"`+id+`"}`), content) - digest, err := st.PublishDigested(t.Context(), staging, id, "") + digest, err := storetest.PublishDigested(t.Context(), st, staging, id) if err != nil { t.Fatalf("PublishDigested: %v", err) } diff --git a/sandboxd/store/store.go b/sandboxd/store/store.go index bb7533286..57c70bacb 100644 --- a/sandboxd/store/store.go +++ b/sandboxd/store/store.go @@ -56,14 +56,16 @@ type S3Config struct { UnsignedAcceptEncoding bool `json:"unsigned_accept_encoding,omitzero"` } +type CommitFunc func(ctx context.Context) error + // Store is one record backend. type Store interface { // Stage returns a writable staging directory whose Publish is atomic. Stage(id string) (string, error) // Publish atomically replaces the record's previous generation with the staged directory; lineage scopes chunk dedup in a content-addressed backend and "" means the record id. Publish(ctx context.Context, staging, id, lineage string) error - // PublishDigested applies Publish semantics and returns the export digest. - PublishDigested(ctx context.Context, staging, id, lineage string) (string, error) + // PrepareDigested runs a Publish up to the step that makes the generation current, and returns the export digest and that step. + PrepareDigested(ctx context.Context, staging, id, lineage string) (string, CommitFunc, error) // Fetch materializes a record's export locally; the caller holds the record lock until the clone ends. Fetch(ctx context.Context, id string) (dir string, meta []byte, digest string, err error) // FetchFile writes one file of a record's export to dst, without fetching the rest of the record; dst is the caller's to delete. diff --git a/sandboxd/store/storetest/storetest.go b/sandboxd/store/storetest/storetest.go index b0d9ac3da..3751bddf1 100644 --- a/sandboxd/store/storetest/storetest.go +++ b/sandboxd/store/storetest/storetest.go @@ -4,6 +4,7 @@ package storetest import ( + "context" "errors" "os" "path/filepath" @@ -98,6 +99,17 @@ func RunContract(t *testing.T, st store.Store) { runLabelsContract(t, st) } +func PublishDigested(ctx context.Context, st store.Store, staging, id string) (string, error) { + digest, commit, err := st.PrepareDigested(ctx, staging, id, "") + if err != nil { + return "", err + } + if err = commit(ctx); err != nil { + return "", err + } + return digest, nil +} + func runFetchFileContract(t *testing.T, st store.Store, id string) { t.Helper() ctx := t.Context() @@ -132,13 +144,19 @@ func runDigestContract(t *testing.T, st store.Store) { if err = os.WriteFile(filepath.Join(staging, store.MetaFile), []byte(`{"id":"`+id+`"}`), 0o600); err != nil { t.Fatalf("write digested meta: %v", err) } - digest, err := st.PublishDigested(ctx, staging, id, "") + digest, commit, err := st.PrepareDigested(ctx, staging, id, "") if err != nil { - t.Fatalf("PublishDigested: %v", err) + t.Fatalf("PrepareDigested: %v", err) } const want = "sha256:ad65b315de70e494c767969e65fc5de65c73846c4ebcdc7051abe08b056637ad" if digest != want { - t.Fatalf("PublishDigested digest %q, want %q", digest, want) + t.Fatalf("PrepareDigested digest %q, want %q", digest, want) + } + if _, err = st.ReadMeta(ctx, id); !errors.Is(err, store.ErrNotFound) { + t.Fatalf("ReadMeta of a prepared record: %v, want store.ErrNotFound until its commit", err) + } + if err = commit(ctx); err != nil { + t.Fatalf("commit: %v", err) } _, _, fetchedDigest, err := st.Fetch(ctx, id) if err != nil { @@ -152,6 +170,27 @@ func runDigestContract(t *testing.T, st store.Store) { t.Errorf("Metas %+v, %v; want the one record with digest %q", recs, err, want) } rejectNonRegularReplacement(t, st, id, want) + staging, err = st.Stage(id) + if err != nil { + t.Fatalf("Stage replacement: %v", err) + } + writeExport(t, staging, "z.bin", "zz") + if err = os.WriteFile(filepath.Join(staging, store.MetaFile), []byte(`{"id":"`+id+`","gen":3}`), 0o600); err != nil { + t.Fatalf("write replacement meta: %v", err) + } + next, commit, err := st.PrepareDigested(ctx, staging, id, "") + if err != nil { + t.Fatalf("PrepareDigested replacement: %v", err) + } + if held, readErr := st.ReadDigest(ctx, id); readErr != nil || held != want { + t.Fatalf("ReadDigest while a replacement is prepared: %q, %v, want the committed %q", held, readErr, want) + } + if err = commit(ctx); err != nil { + t.Fatalf("commit replacement: %v", err) + } + if held, readErr := st.ReadDigest(ctx, id); readErr != nil || held != next { + t.Fatalf("ReadDigest after the replacement commit: %q, %v, want %q", held, readErr, next) + } if err = st.Delete(ctx, id); err != nil { t.Fatalf("Delete digested: %v", err) } @@ -225,8 +264,8 @@ func rejectNonRegularReplacement(t *testing.T, st store.Store, id, wantDigest st if err = os.WriteFile(filepath.Join(staging, store.MetaFile), []byte(`{"id":"`+id+`","gen":2}`), 0o600); err != nil { t.Fatalf("write non-regular meta: %v", err) } - if _, err = st.PublishDigested(t.Context(), staging, id, ""); err == nil { - t.Fatal("PublishDigested accepted a non-regular export entry") + if _, _, err = st.PrepareDigested(t.Context(), staging, id, ""); err == nil { + t.Fatal("PrepareDigested accepted a non-regular export entry") } dir, meta, digest, err := st.Fetch(t.Context(), id) if err != nil {