eric/narratio
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases 15 Wiki Activity

Make remote publish locks generation-safe

Browse Source
This commit is contained in:
Eric Rakestraw
2026-08-10 20:17:54 +00:00
parent 361dbb4ca8
commit 0cf2cbfeb3
22 changed files with 560 additions and 101 deletions
Download Patch File Download Diff File Expand all files Collapse all files

67
internal/app/operator_helpers_test.go
View File

@@ -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)

44
internal/app/operator_locks.go
View File

@@ -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)

11
internal/app/post_publish_cleanup_test.go
View File

@@ -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)
}

89
internal/app/remote_locks.go
View File

@@ -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 {

10
internal/app/restore_discovery_test.go
View File

@@ -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)

8
internal/app/restore_execution_test.go
View File

@@ -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)
}

8
internal/app/runner.go
View File

@@ -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 = &notify.NoopSender{}