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

Remove retired lock authorities

Browse Source
This commit is contained in:
Eric Rakestraw
2026-09-05 13:17:46 +00:00
parent 610eaefb27
commit 7e78c22fec
20 changed files with 73 additions and 723 deletions
Download Patch File Download Diff File Expand all files Collapse all files

24
internal/app/operator_helpers.go
View File

@@ -5,11 +5,14 @@ import (
"flag"
"fmt"
"io"
"os"
"path/filepath"
"strings"
"gitea.maximumdirect.net/eric/narratio/internal/adapters/storage"
"gitea.maximumdirect.net/eric/narratio/internal/artifacts"
"gitea.maximumdirect.net/eric/narratio/internal/config"
"gitea.maximumdirect.net/eric/narratio/internal/fileops"
"gitea.maximumdirect.net/eric/narratio/internal/manifest"
)
@@ -148,3 +151,24 @@ func objectStoreIfConfigured(ctx context.Context, cfg *config.Config) (storage.O
}
return store, nil
}
func writeLocalFile(path string, data []byte, force bool) error {
cleaned := filepath.Clean(strings.TrimSpace(path))
if cleaned == "" || cleaned == "." {
return fmt.Errorf("output path is required")
}
if !force {
if _, err := os.Stat(cleaned); err == nil {
return fmt.Errorf("output file %q already exists; pass --force to overwrite", cleaned)
} else if err != nil && !os.IsNotExist(err) {
return fmt.Errorf("check output file %q: %w", cleaned, err)
}
}
if err := fileops.EnsureWorkspaceDirectory(filepath.Dir(cleaned)); err != nil {
return fmt.Errorf("create output directory: %w", err)
}
if err := os.WriteFile(cleaned, data, fileops.WorkspaceFileMode); err != nil {
return err
}
return os.Chmod(cleaned, fileops.WorkspaceFileMode)
}

192
internal/app/operator_helpers_test.go
View File

@@ -3,13 +3,11 @@ package app
import (
"bytes"
"context"
"errors"
"fmt"
"os"
"path/filepath"
"sort"
"strings"
"sync"
"testing"
"time"
@@ -626,123 +624,6 @@ func TestExecuteLocksListIsSortedAndRejectsMalformedObjects(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 TestLoadRemoteLockStoreAcceptsExactLimitAndRejectsLimitPlusOne(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"},
}
key, err := remoteLocksKey(cfg)
if err != nil {
t.Fatalf("remoteLocksKey() error = %v", err)
}
encoded, err := config.MarshalPublishLockStore(&config.PublishLockStore{})
if err != nil {
t.Fatalf("MarshalPublishLockStore() error = %v", err)
}
exact := append(append([]byte(nil), encoded...), bytes.Repeat([]byte(" "), int(MaxRemoteLockStoreBytes)-len(encoded))...)
store := &storage.FakeBackend{}
store.SeedObject(storage.FakeObject{Key: key, Data: exact, ETag: "lock-generation"})
locks, gotKey, generation, err := loadRemoteLockStore(context.Background(), cfg, store)
if err != nil {
t.Fatalf("loadRemoteLockStore() exact-limit error = %v", err)
}
if locks == nil || gotKey != key || generation != "lock-generation" {
t.Fatalf("loadRemoteLockStore() = (%#v, %q, %q), want decoded locks and opened generation", locks, gotKey, generation)
}
if len(store.Downloads) != 0 || len(store.Reads) != 1 || store.Reads[0].Key != key {
t.Fatalf("lock transfers reads=%#v downloads=%#v, want one direct read", store.Reads, store.Downloads)
}
store.SeedObject(storage.FakeObject{Key: key, Data: append(exact, ' '), ETag: "new-generation"})
_, _, _, err = loadRemoteLockStore(context.Background(), cfg, store)
if err == nil || !strings.Contains(err.Error(), "remote lock control object") || !strings.Contains(err.Error(), key) || !strings.Contains(err.Error(), fmt.Sprint(MaxRemoteLockStoreBytes)) {
t.Fatalf("loadRemoteLockStore() limit-plus-one error = %v, want category, key, and limit", err)
}
}
func TestLoadRemoteLockStoreRejectsMalformedYAML(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"},
}
key, err := remoteLocksKey(cfg)
if err != nil {
t.Fatalf("remoteLocksKey() error = %v", err)
}
store := &storage.FakeBackend{}
store.SeedObject(storage.FakeObject{Key: key, Data: []byte("locks: [\n"), ETag: "lock-generation"})
if _, _, _, err := loadRemoteLockStore(context.Background(), cfg, store); err == nil {
t.Fatal("loadRemoteLockStore() error = nil, want malformed YAML failure")
}
}
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)
@@ -908,39 +789,6 @@ func TestExecuteLocksMutationRejectsSessionIDMismatch(t *testing.T) {
}
}
func TestExecuteLocksDoNotTreatStaticConfigAsAuthority(t *testing.T) {
workspaceRoot := t.TempDir()
pipelinePath, campaignPath, sessionPath := writeValidConfigFiles(t, workspaceRoot)
addStaticPublishLockToPipelineConfig(t, pipelinePath, "narratio.transcript.final_trimmed")
fake := &storage.FakeBackend{}
var storeInitCalls int
restoreAppConfigTestGlobals(t, fake, &storeInitCalls, []string{sessionPath})
var stdout bytes.Buffer
var stderr bytes.Buffer
code := Execute([]string{
"session", "locks", "add", "2026-05-03", "narratio.transcript.final_trimmed",
"--config", pipelinePath,
"--campaign-file", campaignPath,
"--session", sessionPath,
}, &stdout, &stderr)
if code != 0 {
t.Fatalf("locks add exit code = %d, want 0; stderr=%q", code, stderr.String())
}
stdout.Reset()
stderr.Reset()
code = Execute([]string{
"session", "locks", "remove", "2026-05-03", "narratio.transcript.final_trimmed",
"--config", pipelinePath,
"--campaign-file", campaignPath,
"--session", sessionPath,
}, &stdout, &stderr)
if code != 0 {
t.Fatalf("locks remove exit code = %d, want 0; stderr=%q", code, stderr.String())
}
}
func TestExecuteTopLevelLockAndUnlockAreRemoved(t *testing.T) {
tests := []string{"lock", "unlock"}
for _, cmd := range tests {
@@ -1385,11 +1233,12 @@ func TestExecuteSessionValidateReportsPreviousStateFindingAndReturnsFindingError
}
}
func TestExecutePublishLoadsRemoteLocks(t *testing.T) {
func TestExecutePublishIgnoresRetiredSharedLockObject(t *testing.T) {
workspaceRoot := t.TempDir()
pipelinePath, campaignPath, sessionPath := writeValidPublishRunConfigFiles(t, workspaceRoot)
fake := &storage.FakeBackend{}
lockKey := artifacts.S3SessionLocksKey(artifacts.S3SessionPrefix("dnd", "sample-campaign", "2026-05-03"))
sessionPrefix := artifacts.S3SessionPrefix("dnd", "sample-campaign", "2026-05-03")
lockKey := sessionPrefix + "locks.yml"
fake.SeedObject(storage.FakeObject{Key: lockKey, Data: []byte("locks:\n - source: narratio.transcript.final_trimmed\n reason: remote review\n")})
workRoot := filepath.Join(workspaceRoot, "work", "sample-campaign", "2026-05-03")
@@ -1441,9 +1290,18 @@ func TestExecutePublishLoadsRemoteLocks(t *testing.T) {
if storeInitCalls != 1 {
t.Fatalf("object store initialization calls = %d, want 1", storeInitCalls)
}
publishedKey := artifacts.S3PublishedOutputKey(artifacts.S3SessionPrefix("dnd", "sample-campaign", "2026-05-03"), "transcripts/final.trimmed.json")
if _, ok := fake.Objects[publishedKey]; ok {
t.Fatalf("locked published key %q was uploaded", publishedKey)
current, err := artifacts.LoadCurrentState(context.Background(), fake, sessionPrefix, artifacts.CurrentStateValidation{ValidateRunID: true})
if err != nil {
t.Fatalf("LoadCurrentState() error = %v", err)
}
found := false
for _, declaration := range current.Commit.Artifacts {
if declaration.Type == artifacts.RemoteArtifactTypePublishedOutput && declaration.Source == "narratio.transcript.final_trimmed" {
found = true
}
}
if !found {
t.Fatal("retired shared lock object affected publication")
}
}
@@ -1607,23 +1465,3 @@ func writeValidPublishRunConfigFiles(t *testing.T, workspaceRoot string) (string
}
return pipelinePath, campaignPath, sessionPath
}
func addStaticPublishLockToPipelineConfig(t *testing.T, pipelinePath, source string) {
t.Helper()
data, err := os.ReadFile(pipelinePath)
if err != nil {
t.Fatalf("read pipeline: %v", err)
}
updated := strings.Replace(
string(data),
"publish:\n enabled: true\n upload_run: false\n",
"publish:\n enabled: true\n upload_run: false\n locks:\n - source: "+source+"\n reason: static review\n",
1,
)
if updated == string(data) {
t.Fatalf("publish section not found in pipeline config")
}
if err := os.WriteFile(pipelinePath, []byte(updated), 0o644); err != nil {
t.Fatalf("write pipeline: %v", err)
}
}

197
internal/app/remote_locks.go
View File

@@ -1,197 +0,0 @@
package app
import (
"bytes"
"context"
"errors"
"fmt"
"os"
"path/filepath"
"sort"
"strings"
"gitea.maximumdirect.net/eric/narratio/internal/adapters/storage"
"gitea.maximumdirect.net/eric/narratio/internal/artifacts"
"gitea.maximumdirect.net/eric/narratio/internal/config"
"gitea.maximumdirect.net/eric/narratio/internal/fileops"
)
type effectiveLocks struct {
Static []config.PublishLockRule
Remote []config.PublishLockRule
All []config.PublishLockRule
Key string
}
const (
remoteLockMutationAttempts = 4
// MaxRemoteLockStoreBytes bounds the mutable remote publish-lock document.
MaxRemoteLockStoreBytes int64 = 1 << 20
)
func remoteLocksKey(cfg *config.Config) (string, error) {
if cfg == nil || cfg.Pipeline == nil || cfg.Session == nil {
return "", fmt.Errorf("resolved config is required")
}
if cfg.Pipeline.Storage.S3 == nil {
return "", fmt.Errorf("pipeline.storage.s3 configuration is required")
}
sessionPrefix := artifacts.S3SessionPrefix(
cfg.Pipeline.Storage.S3.RootPrefix,
cfg.Session.Campaign,
cfg.Session.SessionID,
)
return artifacts.S3SessionLocksKey(sessionPrefix), nil
}
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
}
if store == nil {
return nil, key, "", fmt.Errorf("remote lock store is required")
}
info, data, err := storage.ReadObjectBounded(ctx, store, key, MaxRemoteLockStoreBytes)
if err != nil {
if errors.Is(err, os.ErrNotExist) {
return &config.PublishLockStore{}, key, "", nil
}
var limitErr *storage.ReadLimitError
if errors.As(err, &limitErr) {
return nil, key, "", fmt.Errorf("read remote lock control object %q with %d-byte limit: %w", key, MaxRemoteLockStoreBytes, err)
}
return nil, key, "", fmt.Errorf("read remote locks %q: %w", key, err)
}
if strings.TrimSpace(info.ETag) == "" {
return nil, key, "", fmt.Errorf("read remote locks %q: object has no generation", key)
}
lockStore, err := config.LoadPublishLockStoreBytes("s3://"+s3BucketName(cfg.Pipeline)+"/"+key, data, cfg.Pipeline.Scriptorium, cfg.Pipeline.Notarius)
if err != nil {
return nil, key, "", err
}
return lockStore, key, info.ETag, nil
}
func loadEffectiveLocks(ctx context.Context, cfg *config.Config, store storage.ObjectStore) (*effectiveLocks, error) {
staticLocks := staticPublishLocks(cfg)
if store == nil {
return &effectiveLocks{
Static: staticLocks,
All: append([]config.PublishLockRule(nil), staticLocks...),
}, nil
}
lockStore, key, _, err := loadRemoteLockStore(ctx, cfg, store)
if err != nil {
return nil, err
}
remoteLocks := append([]config.PublishLockRule(nil), lockStore.Locks...)
return &effectiveLocks{
Static: staticLocks,
Remote: remoteLocks,
All: config.MergePublishLockRules(staticLocks, remoteLocks),
Key: key,
}, nil
}
func staticPublishLocks(cfg *config.Config) []config.PublishLockRule {
if cfg == nil || cfg.Pipeline == nil || cfg.Pipeline.Publish == nil {
return nil
}
return append([]config.PublishLockRule(nil), cfg.Pipeline.Publish.Locks...)
}
func applyEffectiveLocks(cfg *config.Config, locks []config.PublishLockRule) {
if cfg == nil || cfg.Pipeline == nil {
return
}
if cfg.Pipeline.Publish == nil {
cfg.Pipeline.Publish = &config.PublishConfig{}
}
cfg.Pipeline.Publish.Locks = append([]config.PublishLockRule(nil), locks...)
}
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
}
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 {
out := make(map[string]config.PublishLockRule, len(locks))
for _, lock := range locks {
source := strings.TrimSpace(lock.Source)
if source == "" {
continue
}
lock.Source = source
lock.Reason = strings.TrimSpace(lock.Reason)
out[source] = lock
}
return out
}
func lockMapValues(in map[string]config.PublishLockRule) []config.PublishLockRule {
keys := make([]string, 0, len(in))
for key := range in {
keys = append(keys, key)
}
sort.Strings(keys)
out := make([]config.PublishLockRule, 0, len(keys))
for _, key := range keys {
item := in[key]
item.Source = key
item.Reason = strings.TrimSpace(item.Reason)
out = append(out, item)
}
return out
}
func writeLocalFile(path string, data []byte, force bool) error {
cleaned := filepath.Clean(strings.TrimSpace(path))
if cleaned == "" || cleaned == "." {
return fmt.Errorf("output path is required")
}
if !force {
if _, err := os.Stat(cleaned); err == nil {
return fmt.Errorf("output file %q already exists; pass --force to overwrite", cleaned)
} else if err != nil && !os.IsNotExist(err) {
return fmt.Errorf("check output file %q: %w", cleaned, err)
}
}
if err := fileops.EnsureWorkspaceDirectory(filepath.Dir(cleaned)); err != nil {
return fmt.Errorf("create output directory: %w", err)
}
if err := os.WriteFile(cleaned, data, fileops.WorkspaceFileMode); err != nil {
return err
}
return os.Chmod(cleaned, fileops.WorkspaceFileMode)
}