Bound remote control object reads

internal/app/operator_helpers_test.go

@@ -576,6 +576,58 @@ func TestMutateRemoteLockStoreRetainsConcurrentUpdates(t *testing.T) {
 	}
 }
 
+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"}}},

internal/app/remote_locks.go

@@ -5,7 +5,6 @@ import (
 	"context"
 	"errors"
 	"fmt"
-	"io"
 	"os"
 	"path/filepath"
 	"strings"
@@ -23,7 +22,11 @@ type effectiveLocks struct {
 	Key    string
 }
 
-const remoteLockMutationAttempts = 4
+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 {
@@ -48,21 +51,20 @@ func loadRemoteLockStore(ctx context.Context, cfg *config.Config, store storage.
 	if store == nil {
 		return nil, key, "", fmt.Errorf("remote lock store is required")
 	}
-	info, body, err := store.Read(ctx, key)
+	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)
 	}
-	defer body.Close()
 	if strings.TrimSpace(info.ETag) == "" {
 		return nil, key, "", fmt.Errorf("read remote locks %q: object has no generation", key)
 	}
-	data, err := io.ReadAll(body)
-	if err != nil {
-		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

internal/app/restore_commit_test.go

@@ -55,11 +55,11 @@ func TestCommittedRestoreReusesVerifiedManifestCandidate(t *testing.T) {
 	if _, err := executeRestorePlan(context.Background(), cfg, current, plan, nil, fake); err != nil {
 		t.Fatalf("executeRestorePlan() error = %v", err)
 	}
-	if got := fakeDownloadCount(fake, current.CurrentManifestKey); got != 1 {
-		t.Fatalf("manifest downloads = %d, want one discovery transfer reused by restore", got)
+	if got := fakeReadCount(fake, current.CurrentManifestKey); got != 1 {
+		t.Fatalf("manifest reads = %d, want one discovery read reused by restore", got)
 	}
-	if got := fakeDownloadBytes(fake, current.CurrentManifestKey); got != int64(len(current.ManifestData)) {
-		t.Fatalf("manifest bytes transferred = %d, want one verified candidate (%d)", got, len(current.ManifestData))
+	if got := fakeDownloadCount(fake, current.CurrentManifestKey); got != 0 {
+		t.Fatalf("manifest downloads = %d, want no temporary download", got)
 	}
 }
 
@@ -77,8 +77,8 @@ func TestCommittedRestoreRejectsChangedGenerationForVerifiedManifestCandidate(t
 	if err == nil || !strings.Contains(err.Error(), "generation mismatch") {
 		t.Fatalf("executeRestorePlan() error = %v, want generation mismatch", err)
 	}
-	if got := fakeDownloadCount(fake, current.CurrentManifestKey); got != 1 {
-		t.Fatalf("manifest downloads = %d, want no second transfer for rejected candidate", got)
+	if got := fakeReadCount(fake, current.CurrentManifestKey); got != 1 {
+		t.Fatalf("manifest reads = %d, want no second transfer for rejected candidate", got)
 	}
 }
 
@@ -292,11 +292,11 @@ func remoteRestoreArtifact(fake *storage.FakeBackend, artifactType artifacts.Rem
 	}
 }
 
-func fakeDownloadBytes(fake *storage.FakeBackend, key string) int64 {
-	var count int64
-	for _, call := range fake.Downloads {
+func fakeReadCount(fake *storage.FakeBackend, key string) int {
+	count := 0
+	for _, call := range fake.Reads {
 		if call.Key == key {
-			count += call.Bytes
+			count++
 		}
 	}
 	return count

internal/app/restore_discovery_test.go

@@ -152,23 +152,18 @@ func TestDiscoverRemoteCurrentStateUsesCurrentKeysUnderSessionPrefix(t *testing.
 
 	expectedRunKey := fmt.Sprintf("%scurrent/run_id.txt", sessionPrefix)
 	expectedManifestKey := fmt.Sprintf("%scurrent/manifest.json", sessionPrefix)
-	if !containsString(store.existsKeys, expectedRunKey) {
-		t.Fatalf("exists keys = %#v, want run pointer key %q", store.existsKeys, expectedRunKey)
+	if !containsString(store.readKeys, expectedRunKey) {
+		t.Fatalf("read keys = %#v, want run pointer key %q", store.readKeys, expectedRunKey)
 	}
-	if !containsString(store.existsKeys, expectedManifestKey) {
-		t.Fatalf("exists keys = %#v, want manifest key %q", store.existsKeys, expectedManifestKey)
-	}
-	if !containsString(store.downloadKeys, expectedRunKey) {
-		t.Fatalf("download keys = %#v, want run pointer key %q", store.downloadKeys, expectedRunKey)
-	}
-	if !containsString(store.downloadKeys, expectedManifestKey) {
-		t.Fatalf("download keys = %#v, want manifest key %q", store.downloadKeys, expectedManifestKey)
+	if !containsString(store.readKeys, expectedManifestKey) {
+		t.Fatalf("read keys = %#v, want manifest key %q", store.readKeys, expectedManifestKey)
 	}
 }
 
 type captureObjectStore struct {
 	delegate     storage.ObjectStore
 	existsKeys   []string
+	readKeys     []string
 	downloadKeys []string
 }
 
@@ -177,7 +172,7 @@ func (s *captureObjectStore) List(ctx context.Context, prefix string) ([]storage
 }
 
 func (s *captureObjectStore) Read(ctx context.Context, key string) (storage.ObjectInfo, io.ReadCloser, error) {
-	s.downloadKeys = append(s.downloadKeys, key)
+	s.readKeys = append(s.readKeys, key)
 	return s.delegate.Read(ctx, key)
 }
 

internal/app/restore_execution_test.go

@@ -4,8 +4,6 @@ import (
 	"bytes"
 	"context"
 	"encoding/json"
-	"fmt"
-	"io"
 	"os"
 	"path/filepath"
 	"strings"
@@ -360,22 +358,12 @@ func TestExecuteRestoreLockConflictFailsAndWritesNothing(t *testing.T) {
 }
 
 func TestExecuteRestoreInvalidManifestDoesNotCorruptExistingManifest(t *testing.T) {
-	workspaceRoot := t.TempDir()
-	pipelinePath, campaignPath, sessionPath := writeValidConfigFiles(t, workspaceRoot)
-
-	base := &storage.FakeBackend{}
-	cfg, sessionPrefix, manifestKey, _ := seedRestoreCommittedState(t, base, pipelinePath, campaignPath, sessionPath)
-	seedRestoreObject(base, sessionPrefix+"transcripts/full.json", []byte("remote-transcript"))
-
-	toggled := &stagedManifestDownloadStore{
-		delegate:       base,
-		manifestKey:    manifestKey,
-		firstManifest:  restoreManifestJSON(t, cfg.Session.SessionID, cfg.Session.Campaign),
-		secondManifest: []byte("{invalid json"),
-		manifestReads:  0,
-	}
-
-	sessionRoot := artifacts.SessionWorkDirForCampaign(workspaceRoot, cfg.Session.Campaign, cfg.Session.SessionID)
+	cfg := restorePlanConfig(t)
+	store := &storage.FakeBackend{}
+	current := seedCommittedRestoreSnapshot(t, cfg, store, "20260519T010203Z-a1b2c3d4", map[string][]byte{
+		"transcripts/full.json": []byte("remote-transcript"),
+	})
+	sessionRoot := artifacts.SessionWorkDirForCampaign(cfg.Pipeline.Workspace.Root, cfg.Session.Campaign, cfg.Session.SessionID)
 	existing := manifest.New(cfg.Session.SessionID, nowUTC())
 	existing.Campaign = cfg.Session.Campaign
 	existingPath := filepath.Join(sessionRoot, "manifest.json")
@@ -388,25 +376,38 @@ func TestExecuteRestoreInvalidManifestDoesNotCorruptExistingManifest(t *testing.
 		t.Fatalf("read existing local manifest: %v", err)
 	}
 
-	restoreWithStoreAndRealPhases(t, toggled)
-
-	var stdout bytes.Buffer
-	var stderr bytes.Buffer
-	code := Execute([]string{"session", "restore", "2026-05-03", "--config", pipelinePath, "--campaign-file", campaignPath, "--session", sessionPath, "--force"}, &stdout, &stderr)
-	if code == 0 {
-		t.Fatal("exit code = 0, want non-zero")
+	plan, err := buildRestorePlan(context.Background(), cfg, current, store, RestorePlanOptions{Force: true})
+	if err != nil {
+		t.Fatalf("buildRestorePlan() error = %v", err)
 	}
-	if !strings.Contains(stderr.String(), "validate manifest decode") {
-		t.Fatalf("stderr = %q, want manifest validation failure", stderr.String())
+	invalidManifest := []byte("{invalid json")
+	foundManifest := false
+	manifestKey := ""
+	manifestGeneration := ""
+	for index := range plan.Actions {
+		if plan.Actions[index].LocalRelativePath != config.PathManifestFile {
+			continue
+		}
+		foundManifest = true
+		plan.Actions[index].VerifiedContent = invalidManifest
+		plan.Actions[index].SHA256 = restoreCommitSHA256(invalidManifest)
+		plan.Actions[index].Size = int64(len(invalidManifest))
+		manifestKey = plan.Actions[index].RemoteKey
+		manifestGeneration = plan.Actions[index].Generation
+	}
+	if !foundManifest {
+		t.Fatal("restore plan has no manifest action")
+	}
+	store.SeedObject(storage.FakeObject{Key: manifestKey, Data: invalidManifest, ETag: manifestGeneration})
+	report, err := newRestoreReport(current, plan, RestorePlanOptions{Force: true})
+	if err != nil {
+		t.Fatalf("newRestoreReport() error = %v", err)
+	}
+	_, err = executeRestorePlan(context.Background(), cfg, current, plan, report, store)
+	if err == nil || !strings.Contains(err.Error(), "validate manifest decode") {
+		t.Fatalf("executeRestorePlan() error = %v, want manifest validation failure", err)
 	}
 	mustReadEquals(t, filepath.Join(sessionRoot, "transcripts", "full.json"), "remote-transcript")
-	report := mustReadRestoreReport(t, filepath.Join(sessionRoot, "reports", "restore-latest.json"))
-	if report.Status != "failed" {
-		t.Fatalf("report status = %q, want failed", report.Status)
-	}
-	if strings.TrimSpace(report.Error) == "" {
-		t.Fatal("report error is empty, want failure context")
-	}
 	afterData, err := os.ReadFile(existingPath)
 	if err != nil {
 		t.Fatalf("read local manifest after failure: %v", err)
@@ -414,9 +415,6 @@ func TestExecuteRestoreInvalidManifestDoesNotCorruptExistingManifest(t *testing.
 	if string(afterData) != string(existingData) {
 		t.Fatalf("local manifest changed after failed restore; before=%q after=%q", string(existingData), string(afterData))
 	}
-	if _, err := os.Stat(artifacts.SessionRestoreMarkerPathForCampaign(workspaceRoot, cfg.Session.Campaign, cfg.Session.SessionID)); err != nil {
-		t.Fatalf("incomplete restore marker should remain after failed forced restore: %v", err)
-	}
 }
 
 func TestExecuteRestorePlanPathMismatchFails(t *testing.T) {
@@ -562,62 +560,3 @@ func fakeDownloadCount(fake *storage.FakeBackend, key string) int {
 	}
 	return count
 }
-
-type stagedManifestDownloadStore struct {
-	delegate       *storage.FakeBackend
-	manifestKey    string
-	firstManifest  []byte
-	secondManifest []byte
-	manifestReads  int
-}
-
-func (s *stagedManifestDownloadStore) List(ctx context.Context, prefix string) ([]storage.ObjectInfo, error) {
-	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++
-		payload := s.secondManifest
-		if s.manifestReads <= 1 {
-			payload = s.firstManifest
-		}
-		if err := os.MkdirAll(filepath.Dir(localPath), 0o755); err != nil {
-			return fmt.Errorf("download staged manifest: create parent: %w", err)
-		}
-		if err := os.WriteFile(localPath, payload, 0o644); err != nil {
-			return fmt.Errorf("download staged manifest: write local file: %w", err)
-		}
-		return nil
-	}
-	return s.delegate.Download(ctx, key, localPath)
-}
-
-func (s *stagedManifestDownloadStore) DownloadTo(ctx context.Context, key string, destination io.Writer) error {
-	if strings.TrimSpace(key) == strings.TrimSpace(s.manifestKey) {
-		s.manifestReads++
-		payload := s.secondManifest
-		if s.manifestReads <= 1 {
-			payload = s.firstManifest
-		}
-		_, err := destination.Write(payload)
-		return err
-	}
-	return storage.DownloadTo(ctx, s.delegate, key, destination)
-}
-
-func (s *stagedManifestDownloadStore) Upload(ctx context.Context, localPath, key string, opts storage.UploadOptions) (storage.ObjectInfo, error) {
-	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)
-}