diff --git a/docs/internal/stage-publish.md b/docs/internal/stage-publish.md index 19fbb95..6c71846 100644 --- a/docs/internal/stage-publish.md +++ b/docs/internal/stage-publish.md @@ -9,7 +9,8 @@ Upload run/session outputs to object storage and atomically advance remote curre - successful preceding stages from the [canonical stage set](overview.md#pipeline-stage-set) - invocation-scoped run files - resolved publish output rules -- effective publish locks (static + remote merged lock set) +- effective publish locks (static + remote merged lock set), revalidated at the + remote commit boundary - durable previous-session cache files when present ## Outputs @@ -44,6 +45,9 @@ Exact remote placement and the operator workflow belong in - uploads and verifies every declared immutable object and the commit manifest; - updates `current/commit-pointer.json` exactly once, last; and - does not write the legacy `current/manifest.json` or `current/run_id.txt` pair. +- rechecks remote lock state immediately before the pointer update. A newly + committed lock aborts selection, leaving any uploaded immutable attempt + unselected. ## Metadata Signals @@ -66,7 +70,8 @@ Includes counts/lists for: undeclared entries are rejected or ignored before uploads begin. - run-local diagnostics, including Notarius receipt and stderr files, are archived only when recorded by the run manifest. -- publish locks are not overridden by `--force`. +- publish locks are not overridden by `--force`; remote locks are revalidated + immediately before current-state selection. The commit boundary and cleanup gate are normative architecture invariants; see [Architecture](../policy/architecture.md#publish-commit-boundary). diff --git a/docs/internal/storage.md b/docs/internal/storage.md index 640a960..127ea59 100644 --- a/docs/internal/storage.md +++ b/docs/internal/storage.md @@ -12,8 +12,10 @@ operator-selected storage fields and credential mechanisms belong in `storage.ObjectStore` interface: - `List(ctx, prefix)` +- `Read(ctx, key)` returns an object body and the generation observed with it - `Download(ctx, key, localPath)` - `Upload(ctx, localPath, key, opts)` +- `UploadConditional(ctx, source, key, opts, condition)` - `Exists(ctx, key)` Key invariant: @@ -31,8 +33,13 @@ not own discovery, defaults, or configuration validation. - normalizes object keys. - `List` paginates and returns normalized `ObjectInfo`. +- A truncated S3 listing must supply a new, non-empty continuation token; + otherwise listing fails with bucket and prefix context instead of looping. - `Download` writes local files with parent directory creation. - `Upload` streams local file and returns remote metadata. +- `Read` binds a returned body to its S3 ETag. `UploadConditional` maps an ETag + match or absence precondition directly to the provider request and reports a + failed precondition without performing a local check-then-write replacement. - `Exists` maps not-found responses to `false`. ## Invariants diff --git a/docs/operations.md b/docs/operations.md index 42210fe..acc37f0 100644 --- a/docs/operations.md +++ b/docs/operations.md @@ -236,7 +236,14 @@ Effective lock rules: - static and remote locks are merged; - static locks win on source collisions; - locked outputs are intentional skips; -- lock add/remove commands mutate only remote lock state. +- lock add/remove commands mutate only remote lock state through generation-bound + conditional writes. A command retries a bounded number of concurrent + conflicts while its invocation context remains active, so it never replaces a + different lock-document generation; and +- a publish re-reads remote locks immediately before it writes the current + commit pointer. A lock committed before that recheck prevents selecting the + new snapshot, even though its already-uploaded immutable objects may remain + available for a later retry. Examples: diff --git a/docs/policy/architecture.md b/docs/policy/architecture.md index 75d4345..79aa3b7 100644 --- a/docs/policy/architecture.md +++ b/docs/policy/architecture.md @@ -173,7 +173,9 @@ the immutable commit manifest, and finally the current commit pointer. `current/commit-pointer.json` is the sole mutable selector and must be written exactly once, last. Failed, incomplete, skipped, or uncommitted publish attempts must not be presented as current remote state. Publish locks remain authoritative -and are not bypassed by a forced run. +and are not bypassed by a forced run. Mutable remote locks use provider-enforced +generation preconditions and are revalidated immediately before pointer +selection; loss of that check leaves the prior committed snapshot current. Automatic local cleanup is permitted only after a successful publish commit, only when explicitly configured, and only through the path-safety guardrails. diff --git a/docs/roadmap/implementation.md b/docs/roadmap/implementation.md index 391d131..cf4e247 100644 --- a/docs/roadmap/implementation.md +++ b/docs/roadmap/implementation.md @@ -30,7 +30,7 @@ All stages are pending when this plan is created. | 12 | Centralize handled terminal-failure persistence | RSK-001, TST-002, SIM-001, COM-001 | Completed | | 13 | Introduce the immutable remote-commit model and legacy boundary | ARC-003 | Completed | | 14 | Publish through immutable commits and canonical mappings | COR-004, COR-011, DUP-002, TST-004 | Completed | -| 15 | Make remote locks generation-safe and harden pagination | RSK-005, RSK-014 | Pending | +| 15 | Make remote locks generation-safe and harden pagination | RSK-005, RSK-014 | Completed | | 16 | Persist retryable post-commit cleanup state | COR-006, COR-007 | Pending | | 17 | Bind restore/status to a committed snapshot and reject conflicts | COR-008, COR-009, TST-005 | Pending | | 18 | Serialize restore transitions and make restored paths portable | RSK-006, RSK-008 | Pending | diff --git a/internal/adapters/storage/fake.go b/internal/adapters/storage/fake.go index f6b336b..8869b4a 100644 --- a/internal/adapters/storage/fake.go +++ b/internal/adapters/storage/fake.go @@ -1,6 +1,7 @@ package storage import ( + "bytes" "context" "crypto/sha256" "encoding/hex" @@ -10,11 +11,14 @@ import ( "path/filepath" "sort" "strings" + "sync" "time" ) // FakeBackend provides a deterministic in-memory object store for tests. type FakeBackend struct { + mu sync.RWMutex + Objects map[string]FakeObject Uploads []FakeUploadCall Downloads []FakeDownloadCall @@ -50,6 +54,12 @@ type FakeObject struct { // SeedObject inserts or replaces an object in the fake object store. func (f *FakeBackend) SeedObject(obj FakeObject) { + f.mu.Lock() + defer f.mu.Unlock() + f.seedObject(obj) +} + +func (f *FakeBackend) seedObject(obj FakeObject) { if f.Objects == nil { f.Objects = map[string]FakeObject{} } @@ -72,6 +82,8 @@ func (f *FakeBackend) List(ctx context.Context, prefix string) ([]ObjectInfo, er return nil, f.ListErr } + f.mu.RLock() + defer f.mu.RUnlock() normalizedPrefix := normalizeObjectKey(prefix) keys := make([]string, 0, len(f.Objects)) for key := range f.Objects { @@ -94,6 +106,28 @@ func (f *FakeBackend) List(ctx context.Context, prefix string) ([]ObjectInfo, er return out, nil } +// Read returns a stable object body and the generation observed with it. +func (f *FakeBackend) Read(ctx context.Context, key string) (ObjectInfo, io.ReadCloser, error) { + if err := ctx.Err(); err != nil { + return ObjectInfo{}, nil, err + } + if f.DownloadErr != nil { + return ObjectInfo{}, nil, f.DownloadErr + } + normalizedKey := normalizeObjectKey(key) + f.mu.RLock() + obj, ok := f.Objects[normalizedKey] + if ok { + obj.Data = append([]byte(nil), obj.Data...) + obj.Metadata = copyMetadata(obj.Metadata) + } + f.mu.RUnlock() + if !ok { + return ObjectInfo{}, nil, fmt.Errorf("read object %q: %w", normalizedKey, os.ErrNotExist) + } + return ObjectInfo{Key: obj.Key, Size: int64(len(obj.Data)), ETag: obj.ETag, LastModified: obj.LastModified}, io.NopCloser(bytes.NewReader(obj.Data)), nil +} + // DownloadTo writes one object to a caller-owned destination writer. func (f *FakeBackend) DownloadTo(ctx context.Context, key string, destination io.Writer) error { if err := ctx.Err(); err != nil { @@ -106,12 +140,15 @@ func (f *FakeBackend) DownloadTo(ctx context.Context, key string, destination io return fmt.Errorf("download object: destination writer is required") } - obj, ok := f.Objects[normalizeObjectKey(key)] - if !ok { - return fmt.Errorf("download object %q: %w", key, os.ErrNotExist) + _, source, err := f.Read(ctx, key) + if err != nil { + return err } + defer source.Close() + f.mu.Lock() f.Downloads = append(f.Downloads, FakeDownloadCall{Key: normalizeObjectKey(key)}) - if _, err := destination.Write(obj.Data); err != nil { + f.mu.Unlock() + if _, err := io.Copy(destination, source); err != nil { return fmt.Errorf("download object %q: write destination: %w", key, err) } return nil @@ -189,26 +226,71 @@ func (f *FakeBackend) uploadReader(ctx context.Context, source io.Reader, key st ContentType: opts.ContentType, }, } + f.mu.Lock() f.Uploads = append(f.Uploads, call) + f.mu.Unlock() if f.UploadHook != nil { if err := f.UploadHook(call); err != nil { return ObjectInfo{}, err } } - now := time.Now().UTC() - obj := FakeObject{ - Key: normalizedKey, - Data: data, - Metadata: copyMetadata(opts.Metadata), - LastModified: &now, + return f.storeUploadedObject(normalizedKey, data, opts), nil +} + +// UploadConditional atomically checks and replaces one mutable object. +func (f *FakeBackend) UploadConditional(ctx context.Context, source io.Reader, key string, opts UploadOptions, condition WriteCondition) (ObjectInfo, error) { + if err := ctx.Err(); err != nil { + return ObjectInfo{}, err } - f.SeedObject(obj) - return ObjectInfo{ - Key: normalizedKey, - Size: int64(len(data)), - ETag: fakeObjectETag(data), - LastModified: &now, - }, nil + if err := validateWriteCondition(condition); err != nil { + return ObjectInfo{}, err + } + if f.UploadErr != nil { + return ObjectInfo{}, f.UploadErr + } + if source == nil { + return ObjectInfo{}, fmt.Errorf("upload object: source is required") + } + normalizedKey := normalizeObjectKey(key) + if normalizedKey == "" { + return ObjectInfo{}, fmt.Errorf("upload object: key is required") + } + data, err := io.ReadAll(source) + if err != nil { + return ObjectInfo{}, fmt.Errorf("upload object %q: %w", normalizedKey, err) + } + call := FakeUploadCall{Key: normalizedKey, Options: UploadOptions{Metadata: copyMetadata(opts.Metadata), ContentType: opts.ContentType}} + f.mu.Lock() + f.Uploads = append(f.Uploads, call) + f.mu.Unlock() + if f.UploadHook != nil { + if err := f.UploadHook(call); err != nil { + return ObjectInfo{}, err + } + } + f.mu.Lock() + defer f.mu.Unlock() + existing, found := f.Objects[normalizedKey] + if condition.RequireAbsent && found { + return ObjectInfo{}, ErrConditionNotMet + } + if expected := strings.TrimSpace(condition.MatchETag); expected != "" && (!found || existing.ETag != expected) { + return ObjectInfo{}, ErrConditionNotMet + } + return f.storeUploadedObjectLocked(normalizedKey, data, opts), nil +} + +func (f *FakeBackend) storeUploadedObject(key string, data []byte, opts UploadOptions) ObjectInfo { + f.mu.Lock() + defer f.mu.Unlock() + return f.storeUploadedObjectLocked(key, data, opts) +} + +func (f *FakeBackend) storeUploadedObjectLocked(key string, data []byte, opts UploadOptions) ObjectInfo { + now := time.Now().UTC() + obj := FakeObject{Key: key, Data: append([]byte(nil), data...), Metadata: copyMetadata(opts.Metadata), LastModified: &now} + f.seedObject(obj) + return ObjectInfo{Key: key, Size: int64(len(data)), ETag: fakeObjectETag(data), LastModified: &now} } func fakeObjectETag(data []byte) string { @@ -224,7 +306,9 @@ func (f *FakeBackend) Exists(ctx context.Context, key string) (bool, error) { if f.ExistsErr != nil { return false, f.ExistsErr } + f.mu.RLock() _, ok := f.Objects[normalizeObjectKey(key)] + f.mu.RUnlock() return ok, nil } diff --git a/internal/adapters/storage/fake_test.go b/internal/adapters/storage/fake_test.go index 2222098..ed637aa 100644 --- a/internal/adapters/storage/fake_test.go +++ b/internal/adapters/storage/fake_test.go @@ -71,6 +71,21 @@ func TestFakeBackendUploadAndExists(t *testing.T) { } } +func TestFakeBackendConditionalUploadRejectsStaleGeneration(t *testing.T) { + fake := &FakeBackend{} + fake.SeedObject(FakeObject{Key: "locks.yml", Data: []byte("old")}) + old := fake.Objects["locks.yml"].ETag + if _, err := fake.UploadConditional(context.Background(), strings.NewReader("new"), "locks.yml", UploadOptions{}, WriteCondition{MatchETag: old}); err != nil { + t.Fatalf("UploadConditional() error = %v", err) + } + if _, err := fake.UploadConditional(context.Background(), strings.NewReader("lost"), "locks.yml", UploadOptions{}, WriteCondition{MatchETag: old}); !errors.Is(err, ErrConditionNotMet) { + t.Fatalf("UploadConditional() error = %v, want ErrConditionNotMet", err) + } + if got := string(fake.Objects["locks.yml"].Data); got != "new" { + t.Fatalf("locks object = %q, want successful replacement preserved", got) + } +} + func TestFakeBackendObjectErrors(t *testing.T) { fake := &FakeBackend{DownloadErr: errors.New("download fail"), UploadErr: errors.New("upload fail"), ListErr: errors.New("list fail"), ExistsErr: errors.New("exists fail")} diff --git a/internal/adapters/storage/object_store.go b/internal/adapters/storage/object_store.go index 16cbb4b..d8f9863 100644 --- a/internal/adapters/storage/object_store.go +++ b/internal/adapters/storage/object_store.go @@ -2,10 +2,15 @@ package storage import ( "context" + "errors" "io" "time" ) +// ErrConditionNotMet reports that an object changed or already existed before a +// conditional write could be committed. +var ErrConditionNotMet = errors.New("object write condition not met") + // ReaderUploader streams caller-owned, already-opened content to object storage. // Callers retain source-selection and filesystem-confinement policy. type ReaderUploader interface { @@ -19,8 +24,10 @@ type ReaderUploader interface { // infer Narratio session semantics and do not prepend root prefixes. type ObjectStore interface { List(ctx context.Context, prefix string) ([]ObjectInfo, error) + Read(ctx context.Context, key string) (ObjectInfo, io.ReadCloser, error) Download(ctx context.Context, key, localPath string) error Upload(ctx context.Context, localPath, key string, opts UploadOptions) (ObjectInfo, error) + UploadConditional(ctx context.Context, source io.Reader, key string, opts UploadOptions, condition WriteCondition) (ObjectInfo, error) Exists(ctx context.Context, key string) (bool, error) } @@ -37,3 +44,10 @@ type UploadOptions struct { Metadata map[string]string ContentType string } + +// WriteCondition protects a mutable object update against a stale snapshot. +// Exactly one condition is required by UploadConditional. +type WriteCondition struct { + MatchETag string + RequireAbsent bool +} diff --git a/internal/adapters/storage/s3_backend.go b/internal/adapters/storage/s3_backend.go index 3e16543..f7bc64f 100644 --- a/internal/adapters/storage/s3_backend.go +++ b/internal/adapters/storage/s3_backend.go @@ -117,6 +117,7 @@ func (b *S3Backend) List(ctx context.Context, prefix string) ([]ObjectInfo, erro normalizedPrefix := normalizeObjectKey(prefix) out := make([]ObjectInfo, 0) var token *string + seenTokens := map[string]struct{}{} for { resp, err := b.client.ListObjectsV2(ctx, &s3.ListObjectsV2Input{ @@ -142,32 +143,56 @@ func (b *S3Backend) List(ctx context.Context, prefix string) ([]ObjectInfo, erro }) } - if !valueOrFalseBool(resp.IsTruncated) || resp.NextContinuationToken == nil { + if !valueOrFalseBool(resp.IsTruncated) { break } - token = resp.NextContinuationToken + next := strings.TrimSpace(valueOrEmpty(resp.NextContinuationToken)) + if next == "" { + return nil, fmt.Errorf("s3 list objects bucket %q prefix %q: truncated response has an empty continuation token", b.bucket, normalizedPrefix) + } + if _, repeated := seenTokens[next]; repeated { + return nil, fmt.Errorf("s3 list objects bucket %q prefix %q: truncated response repeated continuation token", b.bucket, normalizedPrefix) + } + seenTokens[next] = struct{}{} + token = &next } return out, nil } +// Read retrieves an object together with the generation observed for its body. +func (b *S3Backend) Read(ctx context.Context, key string) (ObjectInfo, io.ReadCloser, error) { + normalizedKey := normalizeObjectKey(key) + resp, err := b.client.GetObject(ctx, &s3.GetObjectInput{Bucket: &b.bucket, Key: &normalizedKey}) + if err != nil { + if isS3NotFound(err) { + return ObjectInfo{}, nil, fmt.Errorf("read object %q: %w", normalizedKey, os.ErrNotExist) + } + return ObjectInfo{}, nil, fmt.Errorf("read object %q: %w", normalizedKey, err) + } + var lastModified *time.Time + if resp.LastModified != nil { + t := *resp.LastModified + lastModified = &t + } + return ObjectInfo{ + Key: normalizedKey, Size: valueOrZeroInt64(resp.ContentLength), + ETag: strings.Trim(valueOrEmpty(resp.ETag), "\""), LastModified: lastModified, + }, resp.Body, nil +} + // DownloadTo retrieves one object into the caller-owned destination writer. func (b *S3Backend) DownloadTo(ctx context.Context, key string, destination io.Writer) error { - normalizedKey := normalizeObjectKey(key) if destination == nil { return fmt.Errorf("download object: destination writer is required") } - - resp, err := b.client.GetObject(ctx, &s3.GetObjectInput{ - Bucket: &b.bucket, - Key: &normalizedKey, - }) + _, body, err := b.Read(ctx, key) if err != nil { - return fmt.Errorf("download object %q: %w", normalizedKey, err) + return fmt.Errorf("download object %q: %w", normalizeObjectKey(key), err) } - defer resp.Body.Close() + defer body.Close() - if _, err := io.Copy(destination, resp.Body); err != nil { - return fmt.Errorf("download object %q: copy body: %w", normalizedKey, err) + if _, err := io.Copy(destination, body); err != nil { + return fmt.Errorf("download object %q: copy body: %w", normalizeObjectKey(key), err) } return nil } @@ -214,15 +239,24 @@ func (b *S3Backend) Upload(ctx context.Context, localPath, key string, opts Uplo if err != nil { return ObjectInfo{}, fmt.Errorf("upload object %q from %q: stat local file: %w", normalizedKey, localPath, err) } - return b.uploadReader(ctx, file, key, opts, stat.Size()) + return b.uploadReader(ctx, file, key, opts, stat.Size(), WriteCondition{}) } // UploadReader sends caller-owned content to key. func (b *S3Backend) UploadReader(ctx context.Context, source io.Reader, key string, opts UploadOptions) (ObjectInfo, error) { - return b.uploadReader(ctx, source, key, opts, 0) + return b.uploadReader(ctx, source, key, opts, 0, WriteCondition{}) } -func (b *S3Backend) uploadReader(ctx context.Context, source io.Reader, key string, opts UploadOptions, size int64) (ObjectInfo, error) { +// UploadConditional uploads a mutable object only when its observed generation +// still matches, or when no object exists yet. +func (b *S3Backend) UploadConditional(ctx context.Context, source io.Reader, key string, opts UploadOptions, condition WriteCondition) (ObjectInfo, error) { + if err := validateWriteCondition(condition); err != nil { + return ObjectInfo{}, err + } + return b.uploadReader(ctx, source, key, opts, 0, condition) +} + +func (b *S3Backend) uploadReader(ctx context.Context, source io.Reader, key string, opts UploadOptions, size int64, condition WriteCondition) (ObjectInfo, error) { normalizedKey := normalizeObjectKey(key) if source == nil { return ObjectInfo{}, fmt.Errorf("upload object: source is required") @@ -241,9 +275,18 @@ func (b *S3Backend) uploadReader(ctx context.Context, source io.Reader, key stri ct := strings.TrimSpace(opts.ContentType) input.ContentType = &ct } + if condition.RequireAbsent { + wildcard := "*" + input.IfNoneMatch = &wildcard + } else if expected := strings.TrimSpace(condition.MatchETag); expected != "" { + input.IfMatch = &expected + } resp, err := b.client.PutObject(ctx, input) if err != nil { + if isS3ConditionalConflict(err) { + return ObjectInfo{}, fmt.Errorf("upload object %q: %w", normalizedKey, ErrConditionNotMet) + } return ObjectInfo{}, fmt.Errorf("upload object %q: %w", normalizedKey, err) } @@ -268,18 +311,43 @@ func (b *S3Backend) Exists(ctx context.Context, key string) (bool, error) { return true, nil } + if isS3NotFound(err) { + return false, nil + } + return false, fmt.Errorf("head object %q: %w", normalizedKey, err) +} + +func validateWriteCondition(condition WriteCondition) error { + if condition.RequireAbsent == (strings.TrimSpace(condition.MatchETag) != "") { + return fmt.Errorf("conditional upload requires exactly one of MatchETag or RequireAbsent") + } + return nil +} + +func isS3NotFound(err error) bool { var notFound *types.NotFound if errors.As(err, ¬Found) { - return false, nil + return true } var apiErr smithy.APIError if errors.As(err, &apiErr) { switch apiErr.ErrorCode() { case "NotFound", "NoSuchKey", "404": - return false, nil + return true } } - return false, fmt.Errorf("head object %q: %w", normalizedKey, err) + return false +} + +func isS3ConditionalConflict(err error) bool { + var apiErr smithy.APIError + if errors.As(err, &apiErr) { + switch apiErr.ErrorCode() { + case "PreconditionFailed", "ConditionalRequestConflict", "412", "409": + return true + } + } + return false } func valueOrEmpty(v *string) string { diff --git a/internal/adapters/storage/s3_backend_test.go b/internal/adapters/storage/s3_backend_test.go index e7781a8..8944e46 100644 --- a/internal/adapters/storage/s3_backend_test.go +++ b/internal/adapters/storage/s3_backend_test.go @@ -2,6 +2,7 @@ package storage import ( "context" + "errors" "io" "os" "path/filepath" @@ -17,8 +18,10 @@ import ( ) type fakeS3API struct { - listOut *s3.ListObjectsV2Output - listErr error + listOut *s3.ListObjectsV2Output + listOutputs []*s3.ListObjectsV2Output + listErr error + listCalls int getBody io.ReadCloser getErr error @@ -28,23 +31,65 @@ type fakeS3API struct { headErr error - lastList *s3.ListObjectsV2Input - lastGet *s3.GetObjectInput - lastPut *s3.PutObjectInput - lastHead *s3.HeadObjectInput + lastList *s3.ListObjectsV2Input + lastLists []*s3.ListObjectsV2Input + lastGet *s3.GetObjectInput + lastPut *s3.PutObjectInput + lastHead *s3.HeadObjectInput } func (f *fakeS3API) ListObjectsV2(_ context.Context, params *s3.ListObjectsV2Input, _ ...func(*s3.Options)) (*s3.ListObjectsV2Output, error) { f.lastList = params + f.lastLists = append(f.lastLists, params) if f.listErr != nil { return nil, f.listErr } + if f.listCalls < len(f.listOutputs) { + out := f.listOutputs[f.listCalls] + f.listCalls++ + return out, nil + } if f.listOut == nil { return &s3.ListObjectsV2Output{}, nil } return f.listOut, nil } +func TestS3BackendListPaginatesAndRejectsNonProgressingTokens(t *testing.T) { + t.Run("multiple pages", func(t *testing.T) { + client := &fakeS3API{listOutputs: []*s3.ListObjectsV2Output{ + {Contents: []types.Object{{Key: strPtr("prefix/a"), Size: int64Ptr(1)}}, IsTruncated: boolPtr(true), NextContinuationToken: strPtr("next")}, + {Contents: []types.Object{{Key: strPtr("prefix/b"), Size: int64Ptr(2)}}, IsTruncated: boolPtr(false)}, + }} + items, err := (&S3Backend{bucket: "bucket-1", client: client}).List(context.Background(), "prefix/") + if err != nil { + t.Fatalf("List() error = %v", err) + } + if len(items) != 2 || items[0].Key != "prefix/a" || items[1].Key != "prefix/b" { + t.Fatalf("List() items = %#v", items) + } + if len(client.lastLists) != 2 || client.lastLists[1].ContinuationToken == nil || *client.lastLists[1].ContinuationToken != "next" { + t.Fatalf("continuation calls = %#v", client.lastLists) + } + }) + + for _, test := range []struct { + name string + outputs []*s3.ListObjectsV2Output + want string + }{ + {name: "empty", outputs: []*s3.ListObjectsV2Output{{IsTruncated: boolPtr(true)}}, want: "empty continuation token"}, + {name: "repeated", outputs: []*s3.ListObjectsV2Output{{IsTruncated: boolPtr(true), NextContinuationToken: strPtr("again")}, {IsTruncated: boolPtr(true), NextContinuationToken: strPtr("again")}}, want: "repeated continuation token"}, + } { + t.Run(test.name, func(t *testing.T) { + _, err := (&S3Backend{bucket: "bucket-1", client: &fakeS3API{listOutputs: test.outputs}}).List(context.Background(), "prefix/") + if err == nil || !strings.Contains(err.Error(), test.want) || !strings.Contains(err.Error(), "bucket-1") || !strings.Contains(err.Error(), "prefix/") { + t.Fatalf("List() error = %v, want contextual %q", err, test.want) + } + }) + } +} + func (f *fakeS3API) GetObject(_ context.Context, params *s3.GetObjectInput, _ ...func(*s3.Options)) (*s3.GetObjectOutput, error) { f.lastGet = params if f.getErr != nil { @@ -160,6 +205,22 @@ func TestS3BackendUploadAndExists(t *testing.T) { } } +func TestS3BackendConditionalUploadUsesProviderPrecondition(t *testing.T) { + client := &fakeS3API{putOut: &s3.PutObjectOutput{ETag: strPtr(`"etag123"`)}} + backend := &S3Backend{bucket: "bucket-1", client: client} + if _, err := backend.UploadConditional(context.Background(), strings.NewReader("payload"), "locks.yml", UploadOptions{}, WriteCondition{MatchETag: "before"}); err != nil { + t.Fatalf("UploadConditional() error = %v", err) + } + if client.lastPut == nil || client.lastPut.IfMatch == nil || *client.lastPut.IfMatch != "before" || client.lastPut.IfNoneMatch != nil { + t.Fatalf("PutObject conditional input = %#v", client.lastPut) + } + client.putErr = &smithy.GenericAPIError{Code: "PreconditionFailed", Message: "changed"} + _, err := backend.UploadConditional(context.Background(), strings.NewReader("payload"), "locks.yml", UploadOptions{}, WriteCondition{RequireAbsent: true}) + if !errors.Is(err, ErrConditionNotMet) { + t.Fatalf("UploadConditional() error = %v, want ErrConditionNotMet", err) + } +} + func TestS3BackendUploadMissingLocalFile(t *testing.T) { backend := &S3Backend{bucket: "bucket-1", client: &fakeS3API{}} _, err := backend.Upload(context.Background(), filepath.Join(t.TempDir(), "missing.txt"), "key.txt", UploadOptions{}) @@ -249,5 +310,6 @@ func TestNewS3BackendFromConfigFallsBackWhenCredentialEnvMissing(t *testing.T) { func strPtr(v string) *string { return &v } func int64Ptr(v int64) *int64 { return &v } +func boolPtr(v bool) *bool { return &v } var _ s3API = (*fakeS3API)(nil) diff --git a/internal/app/operator_helpers_test.go b/internal/app/operator_helpers_test.go index 41be95c..ea5c7d9 100644 --- a/internal/app/operator_helpers_test.go +++ b/internal/app/operator_helpers_test.go @@ -3,10 +3,12 @@ package app import ( "bytes" "context" + "errors" "fmt" "os" "path/filepath" "strings" + "sync" "testing" "time" @@ -522,6 +524,71 @@ func TestExecuteLocksAddListAndRemoveUseRemoteLockStore(t *testing.T) { } } +func TestMutateRemoteLockStoreRetainsConcurrentUpdates(t *testing.T) { + cfg := &config.Config{ + Pipeline: &config.PipelineConfig{Storage: config.StorageConfig{S3: &config.StorageS3Config{Bucket: "bucket", RootPrefix: "root"}}}, + Session: &config.SessionConfig{Campaign: "campaign", SessionID: "session"}, + } + fake := &storage.FakeBackend{} + arrived := make(chan struct{}, 2) + release := make(chan struct{}) + var hookMu sync.Mutex + hookCalls := 0 + fake.UploadHook = func(storage.FakeUploadCall) error { + hookMu.Lock() + hookCalls++ + call := hookCalls + hookMu.Unlock() + if call <= 2 { + arrived <- struct{}{} + <-release + } + return nil + } + + mutate := func(source string) error { + return mutateRemoteLockStore(context.Background(), cfg, fake, func(lockStore *config.PublishLockStore) error { + set := lockSourceSet(lockStore.Locks) + set[source] = config.PublishLockRule{Source: source} + lockStore.Locks = lockMapValues(set) + return nil + }) + } + errs := make(chan error, 2) + go func() { errs <- mutate("narratio.transcript.final") }() + go func() { errs <- mutate("narratio.transcript.final_trimmed") }() + <-arrived + <-arrived + close(release) + if err := <-errs; err != nil { + t.Fatalf("first concurrent mutation error = %v", err) + } + if err := <-errs; err != nil { + t.Fatalf("second concurrent mutation error = %v", err) + } + + locks, _, _, err := loadRemoteLockStore(context.Background(), cfg, fake) + if err != nil { + t.Fatalf("loadRemoteLockStore() error = %v", err) + } + if len(locks.Locks) != 2 || locks.Locks[0].Source != "narratio.transcript.final" || locks.Locks[1].Source != "narratio.transcript.final_trimmed" { + t.Fatalf("remote locks = %#v, want both concurrent updates", locks.Locks) + } +} + +func TestMutateRemoteLockStoreHonorsCancellation(t *testing.T) { + cfg := &config.Config{ + Pipeline: &config.PipelineConfig{Storage: config.StorageConfig{S3: &config.StorageS3Config{Bucket: "bucket", RootPrefix: "root"}}}, + Session: &config.SessionConfig{Campaign: "campaign", SessionID: "session"}, + } + ctx, cancel := context.WithCancel(context.Background()) + cancel() + err := mutateRemoteLockStore(ctx, cfg, &storage.FakeBackend{}, func(*config.PublishLockStore) error { return nil }) + if !errors.Is(err, context.Canceled) { + t.Fatalf("mutateRemoteLockStore() error = %v, want context cancellation", err) + } +} + func TestExecuteLocksAddDuplicateRequiresForce(t *testing.T) { workspaceRoot := t.TempDir() pipelinePath, campaignPath, sessionPath := writeValidConfigFiles(t, workspaceRoot) diff --git a/internal/app/operator_locks.go b/internal/app/operator_locks.go index dd1c8c6..a149597 100644 --- a/internal/app/operator_locks.go +++ b/internal/app/operator_locks.go @@ -73,16 +73,20 @@ func LocksAdd(ctx context.Context, args []string, out io.Writer) error { if _, ok := lockSourceSet(locks.Static)[source]; ok { return fmt.Errorf("locks add: source %q is locked by pipeline config and cannot be modified remotely", source) } - remoteSet := lockSourceSet(locks.Remote) - if _, exists := remoteSet[source]; exists && !force { - return fmt.Errorf("locks add: remote lock for %q already exists; pass --force to update", source) - } - remoteSet[source] = config.PublishLockRule{Source: source, Reason: strings.TrimSpace(reason)} - remoteLocks := lockMapValues(remoteSet) - if _, err := config.ValidatePublishLockRules(remoteLocks, cfg.Pipeline.Scriptorium, cfg.Pipeline.Notarius, "locks"); err != nil { - return fmt.Errorf("locks add: %w", err) - } - if err := uploadRemoteLockStore(ctx, store, locks.Key, &config.PublishLockStore{Locks: remoteLocks}); err != nil { + if err := mutateRemoteLockStore(ctx, cfg, store, func(lockStore *config.PublishLockStore) error { + remoteSet := lockSourceSet(lockStore.Locks) + if _, exists := remoteSet[source]; exists && !force { + return fmt.Errorf("remote lock for %q already exists; pass --force to update", source) + } + remoteSet[source] = config.PublishLockRule{Source: source, Reason: strings.TrimSpace(reason)} + lockStore.Locks = lockMapValues(remoteSet) + normalized, err := config.ValidatePublishLockRules(lockStore.Locks, cfg.Pipeline.Scriptorium, cfg.Pipeline.Notarius, "locks") + if err != nil { + return err + } + lockStore.Locks = normalized + return nil + }); err != nil { return fmt.Errorf("locks add: %w", err) } _, err = fmt.Fprintf(out, "narratio session locks add: locked %s\n", source) @@ -109,16 +113,18 @@ func LocksRemove(ctx context.Context, args []string, out io.Writer) error { if _, err := config.ValidatePublishLockRules([]config.PublishLockRule{{Source: source}}, cfg.Pipeline.Scriptorium, cfg.Pipeline.Notarius, "locks remove"); err != nil { return fmt.Errorf("locks remove: %w", err) } - remoteSet := lockSourceSet(locks.Remote) - if _, ok := remoteSet[source]; !ok { - if _, static := lockSourceSet(locks.Static)[source]; static { - return fmt.Errorf("locks remove: source %q is locked by pipeline config and cannot be unlocked remotely", source) + if err := mutateRemoteLockStore(ctx, cfg, store, func(lockStore *config.PublishLockStore) error { + remoteSet := lockSourceSet(lockStore.Locks) + if _, ok := remoteSet[source]; !ok { + if _, static := lockSourceSet(locks.Static)[source]; static { + return fmt.Errorf("source %q is locked by pipeline config and cannot be unlocked remotely", source) + } + return fmt.Errorf("remote lock for %q does not exist", source) } - return fmt.Errorf("locks remove: remote lock for %q does not exist", source) - } - delete(remoteSet, source) - remoteLocks := lockMapValues(remoteSet) - if err := uploadRemoteLockStore(ctx, store, locks.Key, &config.PublishLockStore{Locks: remoteLocks}); err != nil { + delete(remoteSet, source) + lockStore.Locks = lockMapValues(remoteSet) + return nil + }); err != nil { return fmt.Errorf("locks remove: %w", err) } _, err = fmt.Fprintf(out, "narratio session locks remove: unlocked %s\n", source) diff --git a/internal/app/post_publish_cleanup_test.go b/internal/app/post_publish_cleanup_test.go index 965180d..967fd9d 100644 --- a/internal/app/post_publish_cleanup_test.go +++ b/internal/app/post_publish_cleanup_test.go @@ -394,6 +394,10 @@ func (s *failKeyStore) List(ctx context.Context, prefix string) ([]storage.Objec return s.delegate.List(ctx, prefix) } +func (s *failKeyStore) Read(ctx context.Context, key string) (storage.ObjectInfo, io.ReadCloser, error) { + return s.delegate.Read(ctx, key) +} + func (s *failKeyStore) Download(ctx context.Context, key, localPath string) error { return s.delegate.Download(ctx, key, localPath) } @@ -412,6 +416,13 @@ func (s *failKeyStore) UploadReader(ctx context.Context, source io.Reader, key s return s.delegate.UploadReader(ctx, source, key, opts) } +func (s *failKeyStore) UploadConditional(ctx context.Context, source io.Reader, key string, opts storage.UploadOptions, condition storage.WriteCondition) (storage.ObjectInfo, error) { + if strings.TrimSpace(key) == strings.TrimSpace(s.failKey) { + return storage.ObjectInfo{}, errors.New("forced upload failure") + } + return s.delegate.UploadConditional(ctx, source, key, opts, condition) +} + func (s *failKeyStore) Exists(ctx context.Context, key string) (bool, error) { return s.delegate.Exists(ctx, key) } diff --git a/internal/app/remote_locks.go b/internal/app/remote_locks.go index a251649..9f8dd78 100644 --- a/internal/app/remote_locks.go +++ b/internal/app/remote_locks.go @@ -1,8 +1,11 @@ package app import ( + "bytes" "context" + "errors" "fmt" + "io" "os" "path/filepath" "strings" @@ -20,6 +23,8 @@ type effectiveLocks struct { Key string } +const remoteLockMutationAttempts = 4 + func remoteLocksKey(cfg *config.Config) (string, error) { if cfg == nil || cfg.Pipeline == nil || cfg.Session == nil { return "", fmt.Errorf("resolved config is required") @@ -35,32 +40,34 @@ func remoteLocksKey(cfg *config.Config) (string, error) { return artifacts.S3SessionLocksKey(sessionPrefix), nil } -func loadRemoteLockStore(ctx context.Context, cfg *config.Config, store storage.ObjectStore) (*config.PublishLockStore, string, error) { +func loadRemoteLockStore(ctx context.Context, cfg *config.Config, store storage.ObjectStore) (*config.PublishLockStore, string, string, error) { key, err := remoteLocksKey(cfg) if err != nil { - return nil, "", err + return nil, "", "", err } - exists, err := store.Exists(ctx, key) + if store == nil { + return nil, key, "", fmt.Errorf("remote lock store is required") + } + info, body, err := store.Read(ctx, key) if err != nil { - return nil, key, fmt.Errorf("check remote locks %q: %w", key, err) + if errors.Is(err, os.ErrNotExist) { + return &config.PublishLockStore{}, key, "", nil + } + return nil, key, "", fmt.Errorf("read remote locks %q: %w", key, err) } - if !exists { - return &config.PublishLockStore{}, key, nil + defer body.Close() + if strings.TrimSpace(info.ETag) == "" { + return nil, key, "", fmt.Errorf("read remote locks %q: object has no generation", key) } - tmp, err := storage.DownloadObjectToTemp(ctx, store, key, "narratio-locks-*.yml") + data, err := io.ReadAll(body) if err != nil { - return nil, key, fmt.Errorf("download remote locks %q: %w", key, err) - } - defer func() { _ = os.Remove(tmp) }() - data, err := os.ReadFile(tmp) - if err != nil { - return nil, key, fmt.Errorf("read remote locks %q: %w", key, err) + return nil, key, "", fmt.Errorf("read remote locks %q: %w", key, err) } lockStore, err := config.LoadPublishLockStoreBytes("s3://"+s3BucketName(cfg.Pipeline)+"/"+key, data, cfg.Pipeline.Scriptorium, cfg.Pipeline.Notarius) if err != nil { - return nil, key, err + return nil, key, "", err } - return lockStore, key, nil + return lockStore, key, info.ETag, nil } func loadEffectiveLocks(ctx context.Context, cfg *config.Config, store storage.ObjectStore) (*effectiveLocks, error) { @@ -71,7 +78,7 @@ func loadEffectiveLocks(ctx context.Context, cfg *config.Config, store storage.O All: append([]config.PublishLockRule(nil), staticLocks...), }, nil } - lockStore, key, err := loadRemoteLockStore(ctx, cfg, store) + lockStore, key, _, err := loadRemoteLockStore(ctx, cfg, store) if err != nil { return nil, err } @@ -101,28 +108,38 @@ func applyEffectiveLocks(cfg *config.Config, locks []config.PublishLockRule) { cfg.Pipeline.Publish.Locks = append([]config.PublishLockRule(nil), locks...) } -func uploadRemoteLockStore(ctx context.Context, store storage.ObjectStore, key string, lockStore *config.PublishLockStore) error { - data, err := config.MarshalPublishLockStore(lockStore) - if err != nil { +func mutateRemoteLockStore(ctx context.Context, cfg *config.Config, store storage.ObjectStore, mutate func(*config.PublishLockStore) error) error { + for attempt := 0; attempt < remoteLockMutationAttempts; attempt++ { + if err := ctx.Err(); err != nil { + return err + } + lockStore, key, generation, err := loadRemoteLockStore(ctx, cfg, store) + if err != nil { + return err + } + if err := mutate(lockStore); err != nil { + return err + } + data, err := config.MarshalPublishLockStore(lockStore) + if err != nil { + return err + } + condition := storage.WriteCondition{MatchETag: generation} + if generation == "" { + condition = storage.WriteCondition{RequireAbsent: true} + } + _, err = store.UploadConditional(ctx, bytes.NewReader(data), key, storage.UploadOptions{ContentType: "application/x-yaml; charset=utf-8"}, condition) + if err == nil { + return nil + } + if !errors.Is(err, storage.ErrConditionNotMet) { + return fmt.Errorf("upload remote locks %q: %w", key, err) + } + } + if err := ctx.Err(); err != nil { return err } - tmp, err := os.CreateTemp("", "narratio-locks-upload-*.yml") - if err != nil { - return fmt.Errorf("create lock store temp file: %w", err) - } - tmpPath := tmp.Name() - defer func() { _ = os.Remove(tmpPath) }() - if _, err := tmp.Write(data); err != nil { - _ = tmp.Close() - return fmt.Errorf("write lock store temp file: %w", err) - } - if err := tmp.Close(); err != nil { - return fmt.Errorf("close lock store temp file: %w", err) - } - if _, err := store.Upload(ctx, tmpPath, key, storage.UploadOptions{ContentType: "application/x-yaml; charset=utf-8"}); err != nil { - return fmt.Errorf("upload remote locks %q: %w", key, err) - } - return nil + return fmt.Errorf("update remote locks: concurrent updates prevented a conditional write after %d attempts", remoteLockMutationAttempts) } func lockSourceSet(locks []config.PublishLockRule) map[string]config.PublishLockRule { diff --git a/internal/app/restore_discovery_test.go b/internal/app/restore_discovery_test.go index 1e36005..305b38d 100644 --- a/internal/app/restore_discovery_test.go +++ b/internal/app/restore_discovery_test.go @@ -4,6 +4,7 @@ import ( "context" "encoding/json" "fmt" + "io" "strings" "testing" "time" @@ -175,6 +176,11 @@ func (s *captureObjectStore) List(ctx context.Context, prefix string) ([]storage return s.delegate.List(ctx, prefix) } +func (s *captureObjectStore) Read(ctx context.Context, key string) (storage.ObjectInfo, io.ReadCloser, error) { + s.downloadKeys = append(s.downloadKeys, key) + return s.delegate.Read(ctx, key) +} + func (s *captureObjectStore) Download(ctx context.Context, key, localPath string) error { s.downloadKeys = append(s.downloadKeys, key) return s.delegate.Download(ctx, key, localPath) @@ -184,6 +190,10 @@ func (s *captureObjectStore) Upload(ctx context.Context, localPath, key string, return s.delegate.Upload(ctx, localPath, key, opts) } +func (s *captureObjectStore) UploadConditional(ctx context.Context, source io.Reader, key string, opts storage.UploadOptions, condition storage.WriteCondition) (storage.ObjectInfo, error) { + return s.delegate.UploadConditional(ctx, source, key, opts, condition) +} + func (s *captureObjectStore) Exists(ctx context.Context, key string) (bool, error) { s.existsKeys = append(s.existsKeys, key) return s.delegate.Exists(ctx, key) diff --git a/internal/app/restore_execution_test.go b/internal/app/restore_execution_test.go index dbe4cb8..63f8cff 100644 --- a/internal/app/restore_execution_test.go +++ b/internal/app/restore_execution_test.go @@ -564,6 +564,10 @@ func (s *stagedManifestDownloadStore) List(ctx context.Context, prefix string) ( return s.delegate.List(ctx, prefix) } +func (s *stagedManifestDownloadStore) Read(ctx context.Context, key string) (storage.ObjectInfo, io.ReadCloser, error) { + return s.delegate.Read(ctx, key) +} + func (s *stagedManifestDownloadStore) Download(ctx context.Context, key, localPath string) error { if strings.TrimSpace(key) == strings.TrimSpace(s.manifestKey) { s.manifestReads++ @@ -599,6 +603,10 @@ func (s *stagedManifestDownloadStore) Upload(ctx context.Context, localPath, key return s.delegate.Upload(ctx, localPath, key, opts) } +func (s *stagedManifestDownloadStore) UploadConditional(ctx context.Context, source io.Reader, key string, opts storage.UploadOptions, condition storage.WriteCondition) (storage.ObjectInfo, error) { + return s.delegate.UploadConditional(ctx, source, key, opts, condition) +} + func (s *stagedManifestDownloadStore) Exists(ctx context.Context, key string) (bool, error) { return s.delegate.Exists(ctx, key) } diff --git a/internal/app/runner.go b/internal/app/runner.go index 520c89a..1ee8315 100644 --- a/internal/app/runner.go +++ b/internal/app/runner.go @@ -210,6 +210,14 @@ func executeStages(ctx context.Context, cfg *config.Config, stages []stage.Stage ) } applyEffectiveLocks(env.Config, locks.All) + staticLocks := append([]config.PublishLockRule(nil), locks.Static...) + env.RevalidatePublishLocks = func(recheckCtx context.Context) ([]config.PublishLockRule, error) { + remote, _, _, err := loadRemoteLockStore(recheckCtx, env.Config, env.ObjectStore) + if err != nil { + return nil, err + } + return config.MergePublishLockRules(staticLocks, remote.Locks), nil + } } if env.Notifier == nil { env.Notifier = ¬ify.NoopSender{} diff --git a/internal/stage/analyze_test.go b/internal/stage/analyze_test.go index 983783a..2cf2788 100644 --- a/internal/stage/analyze_test.go +++ b/internal/stage/analyze_test.go @@ -3,6 +3,7 @@ package stage import ( "context" "errors" + "io" "os" "path/filepath" "sort" @@ -1622,6 +1623,11 @@ func (s *analyzeObjectStoreTracker) List(context.Context, string) ([]storage.Obj return nil, errors.New("unexpected object store list call") } +func (s *analyzeObjectStoreTracker) Read(context.Context, string) (storage.ObjectInfo, io.ReadCloser, error) { + s.called = true + return storage.ObjectInfo{}, nil, errors.New("unexpected object store read call") +} + func (s *analyzeObjectStoreTracker) Download(context.Context, string, string) error { s.called = true return errors.New("unexpected object store download call") @@ -1632,6 +1638,11 @@ func (s *analyzeObjectStoreTracker) Upload(context.Context, string, string, stor return storage.ObjectInfo{}, errors.New("unexpected object store upload call") } +func (s *analyzeObjectStoreTracker) UploadConditional(context.Context, io.Reader, string, storage.UploadOptions, storage.WriteCondition) (storage.ObjectInfo, error) { + s.called = true + return storage.ObjectInfo{}, errors.New("unexpected conditional object store upload call") +} + func (s *analyzeObjectStoreTracker) Exists(context.Context, string) (bool, error) { s.called = true return false, errors.New("unexpected object store exists call") diff --git a/internal/stage/prepare_previous_test.go b/internal/stage/prepare_previous_test.go index 8deba5e..f10b29e 100644 --- a/internal/stage/prepare_previous_test.go +++ b/internal/stage/prepare_previous_test.go @@ -387,6 +387,11 @@ func (s *preparePreviousCaptureStore) List(ctx context.Context, prefix string) ( return s.delegate.List(ctx, prefix) } +func (s *preparePreviousCaptureStore) Read(ctx context.Context, key string) (storage.ObjectInfo, io.ReadCloser, error) { + s.downloadKeys = append(s.downloadKeys, key) + return s.delegate.Read(ctx, key) +} + func (s *preparePreviousCaptureStore) Download(ctx context.Context, key, localPath string) error { s.downloadKeys = append(s.downloadKeys, key) return s.delegate.Download(ctx, key, localPath) @@ -401,6 +406,10 @@ func (s *preparePreviousCaptureStore) Upload(ctx context.Context, localPath, key return s.delegate.Upload(ctx, localPath, key, opts) } +func (s *preparePreviousCaptureStore) UploadConditional(ctx context.Context, source io.Reader, key string, opts storage.UploadOptions, condition storage.WriteCondition) (storage.ObjectInfo, error) { + return s.delegate.UploadConditional(ctx, source, key, opts, condition) +} + func (s *preparePreviousCaptureStore) Exists(ctx context.Context, key string) (bool, error) { s.existsKeys = append(s.existsKeys, key) return s.delegate.Exists(ctx, key) diff --git a/internal/stage/publish.go b/internal/stage/publish.go index 4a28f3f..e8665ea 100644 --- a/internal/stage/publish.go +++ b/internal/stage/publish.go @@ -231,6 +231,9 @@ func (publishStage) Run(ctx context.Context, env *Env, m *manifest.Manifest) (*S if err != nil { return nil, fmt.Errorf("publish: upload immutable commit %q: %w", commitKey, err) } + if err := revalidatePublishCommitLocks(ctx, env, publishOutputs); err != nil { + return nil, fmt.Errorf("publish: revalidate locks before current commit selection: %w", err) + } pointerData, err := artifacts.EncodeCurrentCommitPointer(artifacts.CurrentCommitPointer{ FormatVersion: artifacts.RemoteCommitFormatVersion, Campaign: commit.Campaign, @@ -271,6 +274,23 @@ func (publishStage) Run(ctx context.Context, env *Env, m *manifest.Manifest) (*S }, nil } +func revalidatePublishCommitLocks(ctx context.Context, env *Env, outputs []publishOutput) error { + if env == nil || env.RevalidatePublishLocks == nil || len(outputs) == 0 { + return nil + } + locks, err := env.RevalidatePublishLocks(ctx) + if err != nil { + return err + } + lockSet := publishLockSet(locks) + for _, output := range outputs { + if lock, locked := lockSet[output.Source]; locked { + return fmt.Errorf("source %q is locked: %s", output.Source, strings.TrimSpace(lock.Reason)) + } + } + return nil +} + type publishOutput struct { Source string Dest string diff --git a/internal/stage/publish_test.go b/internal/stage/publish_test.go index 279d2d9..2f4f72e 100644 --- a/internal/stage/publish_test.go +++ b/internal/stage/publish_test.go @@ -885,6 +885,30 @@ func TestPublishRejectsConflictingImmutableObject(t *testing.T) { } } +func TestPublishDoesNotSelectCommitWhenLockAppearsAtCommitPoint(t *testing.T) { + env, m, _ := publishFixture(t) + fake := env.ObjectStore.(*storage.FakeBackend) + oldRunID := "20260515T010203Z-a1b2c3d4" + seedCommittedCurrentState(t, fake, m, oldRunID) + pointerKey := artifacts.S3CurrentCommitPointerKey(m.S3SessionPrefix) + priorPointer := append([]byte(nil), fake.Objects[pointerKey].Data...) + env.RevalidatePublishLocks = func(context.Context) ([]config.PublishLockRule, error) { + return []config.PublishLockRule{{Source: "narratio.transcript.final_trimmed", Reason: "manual review"}}, nil + } + + _, err := (publishStage{}).Run(context.Background(), env, m) + if err == nil || !strings.Contains(err.Error(), "revalidate locks") || !strings.Contains(err.Error(), "manual review") { + t.Fatalf("Run() error = %v, want commit-point lock failure", err) + } + if got := fake.Objects[pointerKey].Data; !reflect.DeepEqual(got, priorPointer) { + t.Fatalf("current pointer changed after lock appeared: %q", got) + } + state, err := artifacts.LoadCurrentState(context.Background(), fake, m.S3SessionPrefix, artifacts.CurrentStateValidation{ValidateRunID: true}) + if err != nil || state.RunID != oldRunID { + t.Fatalf("current state after lock loss = %#v, %v; want old run %q", state, err, oldRunID) + } +} + func TestPublishRejectsAmbiguousOrCollidingOutputMappings(t *testing.T) { tests := []struct { name string diff --git a/internal/stage/stage.go b/internal/stage/stage.go index bff6a4f..2577d2f 100644 --- a/internal/stage/stage.go +++ b/internal/stage/stage.go @@ -33,6 +33,10 @@ type Env struct { Scriptorium scriptorium.Runner ObjectStore storage.ObjectStore Notifier notify.Sender + + // RevalidatePublishLocks returns the effective lock set immediately before a + // publish commit selects a new remote snapshot. + RevalidatePublishLocks func(context.Context) ([]config.PublishLockRule, error) } // IODecl declares the intended input/output artifact kinds for a stage.