Serialize restore recovery and rebase manifest paths
This commit is contained in:
@@ -8,8 +8,8 @@ reporting flow in `internal/app`. User invocation belongs in
|
|||||||
physical restore scope belong in
|
physical restore scope belong in
|
||||||
[Operations](../operations.md#restore-workflow).
|
[Operations](../operations.md#restore-workflow).
|
||||||
|
|
||||||
Restore is split into explicit phases so remote authority, local conflict
|
Restore separates remote authority, local conflict policy, and filesystem
|
||||||
policy, and filesystem mutation can be tested independently.
|
mutation so each remains testable independently.
|
||||||
|
|
||||||
## Discovery Contract
|
## Discovery Contract
|
||||||
|
|
||||||
@@ -40,6 +40,9 @@ Planner behavior:
|
|||||||
- force converts differing eligible regular files from conflicts to downloads;
|
- force converts differing eligible regular files from conflicts to downloads;
|
||||||
directories and other non-regular targets remain conflicts.
|
directories and other non-regular targets remain conflicts.
|
||||||
|
|
||||||
|
For a non-dry-run restore, planning/classification happens only after acquiring
|
||||||
|
the session lock. Runner manifest/reuse checks acquire that same lock first.
|
||||||
|
|
||||||
Previous-cache files are planned separately through `previouscache.BuildPlan`
|
Previous-cache files are planned separately through `previouscache.BuildPlan`
|
||||||
when configured previous-session requirements exist.
|
when configured previous-session requirements exist.
|
||||||
|
|
||||||
@@ -54,6 +57,13 @@ Execution order and safety:
|
|||||||
- each committed object is verified against its declared checksum, size, and
|
- each committed object is verified against its declared checksum, size, and
|
||||||
generation before installation;
|
generation before installation;
|
||||||
- failed installs do not roll back files already written in the same execution.
|
- failed installs do not roll back files already written in the same execution.
|
||||||
|
- a durable `.restore-incomplete.json` marker is written before installation.
|
||||||
|
It blocks runners until a restore retry completes all verified installs and
|
||||||
|
the local manifest replacement, at which point it is removed.
|
||||||
|
- restored manifest local references are rebased beneath the selected local
|
||||||
|
session root. Unsafe relative references and producer-machine absolute paths
|
||||||
|
outside the manifest's producer session root are rejected; producer-local
|
||||||
|
spool/cache and cleanup locations are not restored as authority.
|
||||||
|
|
||||||
Audio restore path:
|
Audio restore path:
|
||||||
|
|
||||||
|
|||||||
@@ -283,6 +283,12 @@ Optional:
|
|||||||
|
|
||||||
Restore writes an execution report at `reports/restore-latest.json`.
|
Restore writes an execution report at `reports/restore-latest.json`.
|
||||||
|
|
||||||
|
If restore fails after beginning installation, it leaves a durable
|
||||||
|
`.restore-incomplete.json` marker in the session root. Pipeline runs will stop
|
||||||
|
until you rerun the same restore command and it completes. Restore intentionally
|
||||||
|
does not try to roll back files already installed; retrying the selected remote
|
||||||
|
snapshot is the recovery procedure.
|
||||||
|
|
||||||
## Local State Layout
|
## Local State Layout
|
||||||
|
|
||||||
Session root:
|
Session root:
|
||||||
|
|||||||
@@ -119,6 +119,12 @@ install the validated session manifest after other restored durable files. The
|
|||||||
physical workflow and recovery procedures belong in
|
physical workflow and recovery procedures belong in
|
||||||
[Operations](../operations.md).
|
[Operations](../operations.md).
|
||||||
|
|
||||||
|
Restore and runner transitions for one session use the same local lock. A
|
||||||
|
durable incomplete-restore marker blocks runner reuse after a partial restore;
|
||||||
|
safe retry, rather than rollback of arbitrary local effects, is the recovery
|
||||||
|
mechanism. Restored manifest-local references must be confined to the selected
|
||||||
|
local session root, never trusted as producer-machine absolute paths.
|
||||||
|
|
||||||
For the immutable remote-commit protocol, a restore or status operation binds
|
For the immutable remote-commit protocol, a restore or status operation binds
|
||||||
to one pointer-selected commit and only its declared object identities. A force
|
to one pointer-selected commit and only its declared object identities. A force
|
||||||
flag may replace an eligible regular managed file, but never turns a directory
|
flag may replace an eligible regular managed file, but never turns a directory
|
||||||
|
|||||||
@@ -33,7 +33,7 @@ All stages are pending when this plan is created.
|
|||||||
| 15 | Make remote locks generation-safe and harden pagination | RSK-005, RSK-014 | Completed |
|
| 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 | Completed |
|
| 16 | Persist retryable post-commit cleanup state | COR-006, COR-007 | Completed |
|
||||||
| 17 | Bind restore/status to a committed snapshot and reject conflicts | COR-008, COR-009, TST-005 | Completed |
|
| 17 | Bind restore/status to a committed snapshot and reject conflicts | COR-008, COR-009, TST-005 | Completed |
|
||||||
| 18 | Serialize restore transitions and make restored paths portable | RSK-006, RSK-008 | Pending |
|
| 18 | Serialize restore transitions and make restored paths portable | RSK-006, RSK-008 | Completed |
|
||||||
| 19 | Bind audio cache reuse to remote object identity | RSK-007 | Pending |
|
| 19 | Bind audio cache reuse to remote object identity | RSK-007 | Pending |
|
||||||
| 20 | Unify previous-source readiness and eliminate duplicate transfers | COR-010, EFF-001 | Pending |
|
| 20 | Unify previous-source readiness and eliminate duplicate transfers | COR-010, EFF-001 | Pending |
|
||||||
| 21 | Tighten configuration parsing, values, and expectations | COR-012–COR-015, TST-011, TST-014 | Pending |
|
| 21 | Tighten configuration parsing, values, and expectations | COR-012–COR-015, TST-011, TST-014 | Pending |
|
||||||
|
|||||||
@@ -69,23 +69,23 @@ func Restore(ctx context.Context, args []string, out io.Writer) (resultErr error
|
|||||||
if err != nil {
|
if err != nil {
|
||||||
return fmt.Errorf("restore: %w", err)
|
return fmt.Errorf("restore: %w", err)
|
||||||
}
|
}
|
||||||
plan, err := buildRestorePlanFn(ctx, cfg, current, objectStore, RestorePlanOptions{
|
|
||||||
IncludeAudio: includeAudio,
|
|
||||||
Force: force,
|
|
||||||
DryRun: dryRun,
|
|
||||||
})
|
|
||||||
if err != nil {
|
|
||||||
return fmt.Errorf("restore: %w", err)
|
|
||||||
}
|
|
||||||
report, err := newRestoreReport(current, plan, RestorePlanOptions{
|
|
||||||
IncludeAudio: includeAudio,
|
|
||||||
Force: force,
|
|
||||||
DryRun: dryRun,
|
|
||||||
})
|
|
||||||
if err != nil {
|
|
||||||
return fmt.Errorf("restore: %w", err)
|
|
||||||
}
|
|
||||||
if dryRun {
|
if dryRun {
|
||||||
|
plan, err := buildRestorePlanFn(ctx, cfg, current, objectStore, RestorePlanOptions{
|
||||||
|
IncludeAudio: includeAudio,
|
||||||
|
Force: force,
|
||||||
|
DryRun: true,
|
||||||
|
})
|
||||||
|
if err != nil {
|
||||||
|
return fmt.Errorf("restore: %w", err)
|
||||||
|
}
|
||||||
|
report, err := newRestoreReport(current, plan, RestorePlanOptions{
|
||||||
|
IncludeAudio: includeAudio,
|
||||||
|
Force: force,
|
||||||
|
DryRun: true,
|
||||||
|
})
|
||||||
|
if err != nil {
|
||||||
|
return fmt.Errorf("restore: %w", err)
|
||||||
|
}
|
||||||
if err := writeRestoreDryRunSummary(out, report); err != nil {
|
if err := writeRestoreDryRunSummary(out, report); err != nil {
|
||||||
return fmt.Errorf("restore: write plan output: %w", err)
|
return fmt.Errorf("restore: write plan output: %w", err)
|
||||||
}
|
}
|
||||||
@@ -110,6 +110,24 @@ func Restore(ctx context.Context, args []string, out io.Writer) (resultErr error
|
|||||||
}
|
}
|
||||||
}()
|
}()
|
||||||
|
|
||||||
|
// Classification and installation share the same transition lock as a
|
||||||
|
// runner. This prevents a runner from making reuse decisions against state
|
||||||
|
// that restore is about to replace.
|
||||||
|
plan, err := buildRestorePlanFn(ctx, cfg, current, objectStore, RestorePlanOptions{
|
||||||
|
IncludeAudio: includeAudio,
|
||||||
|
Force: force,
|
||||||
|
})
|
||||||
|
if err != nil {
|
||||||
|
return fmt.Errorf("restore: %w", err)
|
||||||
|
}
|
||||||
|
report, err := newRestoreReport(current, plan, RestorePlanOptions{
|
||||||
|
IncludeAudio: includeAudio,
|
||||||
|
Force: force,
|
||||||
|
})
|
||||||
|
if err != nil {
|
||||||
|
return fmt.Errorf("restore: %w", err)
|
||||||
|
}
|
||||||
|
|
||||||
if plan.ConflictCount > 0 {
|
if plan.ConflictCount > 0 {
|
||||||
report.setFailed(fmt.Errorf("conflict: %d conflicting path(s)", plan.ConflictCount))
|
report.setFailed(fmt.Errorf("conflict: %d conflicting path(s)", plan.ConflictCount))
|
||||||
if _, reportErr := persistRestoreReport(artifactStore, cfg, report); reportErr != nil {
|
if _, reportErr := persistRestoreReport(artifactStore, cfg, report); reportErr != nil {
|
||||||
@@ -123,6 +141,9 @@ func Restore(ctx context.Context, args []string, out io.Writer) (resultErr error
|
|||||||
plan.ConflictCount,
|
plan.ConflictCount,
|
||||||
)
|
)
|
||||||
}
|
}
|
||||||
|
if err := writeRestoreMarker(artifactStore.SessionPathsFor(cfg.Session.Campaign, cfg.Session.SessionID)); err != nil {
|
||||||
|
return fmt.Errorf("restore: mark incomplete restore: %w", err)
|
||||||
|
}
|
||||||
|
|
||||||
result, err := executeRestorePlanFn(ctx, cfg, current, plan, report, objectStore)
|
result, err := executeRestorePlanFn(ctx, cfg, current, plan, report, objectStore)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
@@ -134,6 +155,9 @@ func Restore(ctx context.Context, args []string, out io.Writer) (resultErr error
|
|||||||
}
|
}
|
||||||
report.Execution.Downloaded = result.DownloadedCount
|
report.Execution.Downloaded = result.DownloadedCount
|
||||||
report.setSucceeded()
|
report.setSucceeded()
|
||||||
|
if err := clearRestoreMarker(artifactStore.SessionPathsFor(cfg.Session.Campaign, cfg.Session.SessionID)); err != nil {
|
||||||
|
return fmt.Errorf("restore: clear incomplete restore marker: %w", err)
|
||||||
|
}
|
||||||
if _, err := persistRestoreReport(artifactStore, cfg, report); err != nil {
|
if _, err := persistRestoreReport(artifactStore, cfg, report); err != nil {
|
||||||
return fmt.Errorf("restore: write report: %w", err)
|
return fmt.Errorf("restore: write report: %w", err)
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -114,19 +114,28 @@ func executeRestoreDownloadAction(
|
|||||||
return err
|
return err
|
||||||
}
|
}
|
||||||
|
|
||||||
if action.LocalRelativePath == config.PathManifestFile {
|
if isRestoreManifest(action.LocalRelativePath) {
|
||||||
file, err := temporary.Open()
|
file, err := temporary.Open()
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return fmt.Errorf("open restored manifest: %w", err)
|
return fmt.Errorf("open restored manifest: %w", err)
|
||||||
}
|
}
|
||||||
err = validateRestoredManifest(ctx, cfg, current, file)
|
destinationRoot := sessionRoot
|
||||||
|
requireCurrentIdentity := action.LocalRelativePath == config.PathManifestFile
|
||||||
|
if !requireCurrentIdentity {
|
||||||
|
destinationRoot = filepath.Dir(safeLocalPath)
|
||||||
|
}
|
||||||
|
restored, prepareErr := prepareRestoredManifest(ctx, cfg, current, file, destinationRoot, requireCurrentIdentity)
|
||||||
closeErr := file.Close()
|
closeErr := file.Close()
|
||||||
if err != nil {
|
if prepareErr != nil {
|
||||||
return err
|
return prepareErr
|
||||||
}
|
}
|
||||||
if closeErr != nil {
|
if closeErr != nil {
|
||||||
return fmt.Errorf("close restored manifest: %w", closeErr)
|
return fmt.Errorf("close restored manifest: %w", closeErr)
|
||||||
}
|
}
|
||||||
|
if err := (&manifest.LocalStore{}).Save(ctx, safeLocalPath, restored); err != nil {
|
||||||
|
return fmt.Errorf("install rebased manifest atomically: %w", err)
|
||||||
|
}
|
||||||
|
return nil
|
||||||
}
|
}
|
||||||
|
|
||||||
if err := temporary.Install(filepath.Base(safeLocalPath), fileops.WorkspaceFileMode); err != nil {
|
if err := temporary.Install(filepath.Base(safeLocalPath), fileops.WorkspaceFileMode); err != nil {
|
||||||
@@ -136,6 +145,11 @@ func executeRestoreDownloadAction(
|
|||||||
return nil
|
return nil
|
||||||
}
|
}
|
||||||
|
|
||||||
|
func isRestoreManifest(relativePath string) bool {
|
||||||
|
clean := filepath.ToSlash(filepath.Clean(strings.TrimSpace(relativePath)))
|
||||||
|
return clean == config.PathManifestFile || clean == config.PathPreviousDirSegment+"/"+config.PathManifestFile
|
||||||
|
}
|
||||||
|
|
||||||
func verifyRestoredObject(ctx context.Context, store storage.ObjectStore, action RestoreAction, temporary *fileops.DownloadedTempFile) error {
|
func verifyRestoredObject(ctx context.Context, store storage.ObjectStore, action RestoreAction, temporary *fileops.DownloadedTempFile) error {
|
||||||
if strings.TrimSpace(action.SHA256) == "" && strings.TrimSpace(action.Generation) == "" {
|
if strings.TrimSpace(action.SHA256) == "" && strings.TrimSpace(action.Generation) == "" {
|
||||||
return nil
|
return nil
|
||||||
@@ -217,38 +231,3 @@ func executeRestoreAudioAction(
|
|||||||
}
|
}
|
||||||
return nil
|
return nil
|
||||||
}
|
}
|
||||||
|
|
||||||
func validateRestoredManifest(ctx context.Context, cfg *config.Config, current *RemoteCurrentState, source io.Reader) error {
|
|
||||||
manifestStore := &manifest.LocalStore{}
|
|
||||||
m, err := manifestStore.LoadReader(ctx, source)
|
|
||||||
if err != nil {
|
|
||||||
return fmt.Errorf("validate manifest decode: %w", err)
|
|
||||||
}
|
|
||||||
|
|
||||||
requestedSession := strings.TrimSpace(cfg.Session.SessionID)
|
|
||||||
requestedCampaign := strings.TrimSpace(cfg.Session.Campaign)
|
|
||||||
manifestSession := strings.TrimSpace(m.SessionID)
|
|
||||||
manifestCampaign := strings.TrimSpace(m.Campaign)
|
|
||||||
if manifestSession != requestedSession {
|
|
||||||
return fmt.Errorf("manifest session_id %q does not match requested session_id %q", manifestSession, requestedSession)
|
|
||||||
}
|
|
||||||
if manifestCampaign == "" {
|
|
||||||
return fmt.Errorf("manifest campaign is required")
|
|
||||||
}
|
|
||||||
if manifestCampaign != requestedCampaign {
|
|
||||||
return fmt.Errorf("manifest campaign %q does not match requested campaign %q", manifestCampaign, requestedCampaign)
|
|
||||||
}
|
|
||||||
if current != nil {
|
|
||||||
if expected := strings.TrimSpace(current.SessionID); expected != "" && manifestSession != expected {
|
|
||||||
return fmt.Errorf("manifest session_id %q does not match discovered session_id %q", manifestSession, expected)
|
|
||||||
}
|
|
||||||
if expected := strings.TrimSpace(current.Campaign); expected != "" && manifestCampaign != expected {
|
|
||||||
return fmt.Errorf("manifest campaign %q does not match discovered campaign %q", manifestCampaign, expected)
|
|
||||||
}
|
|
||||||
if expected := strings.TrimSpace(current.RunID); expected != "" && strings.TrimSpace(m.RunID) != expected {
|
|
||||||
return fmt.Errorf("manifest run_id %q does not match discovered run_id %q", strings.TrimSpace(m.RunID), expected)
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
return nil
|
|
||||||
}
|
|
||||||
|
|||||||
@@ -76,11 +76,13 @@ func TestExecuteRestoreRoundTripsPublishedExtractionAndManifestMetadata(t *testi
|
|||||||
remoteManifest := manifest.New(cfg.Session.SessionID, time.Now().UTC())
|
remoteManifest := manifest.New(cfg.Session.SessionID, time.Now().UTC())
|
||||||
remoteManifest.Campaign = cfg.Session.Campaign
|
remoteManifest.Campaign = cfg.Session.Campaign
|
||||||
remoteManifest.RunID = "20260519T010203Z-a1b2c3d4"
|
remoteManifest.RunID = "20260519T010203Z-a1b2c3d4"
|
||||||
|
producerRoot := "/prior/workspace/work/sample-campaign/2026-05-03"
|
||||||
|
remoteManifest.LocalWorkDir = filepath.Join(producerRoot, "runs", remoteManifest.RunID)
|
||||||
remoteManifest.Stages["extract"] = &manifest.StageRecord{
|
remoteManifest.Stages["extract"] = &manifest.StageRecord{
|
||||||
Name: "extract", Status: manifest.StatusSucceeded,
|
Name: "extract", Status: manifest.StatusSucceeded,
|
||||||
Outputs: []manifest.ArtifactRecord{{
|
Outputs: []manifest.ArtifactRecord{{
|
||||||
Kind: "notarius_lane", SourceID: artifacts.ExtractionArtifactSourceID("encounters"),
|
Kind: "notarius_lane", SourceID: artifacts.ExtractionArtifactSourceID("encounters"),
|
||||||
LocalPath: "/prior/workspace/artifacts/notarius/extract-run-1/lanes/encounters.json",
|
LocalPath: filepath.Join(producerRoot, "artifacts", "encounters.json"),
|
||||||
Contract: &artifactmodel.ContractMetadata{
|
Contract: &artifactmodel.ContractMetadata{
|
||||||
MediaType: "application/json", SchemaID: "encounters", SchemaVersion: "1",
|
MediaType: "application/json", SchemaID: "encounters", SchemaVersion: "1",
|
||||||
},
|
},
|
||||||
@@ -112,6 +114,9 @@ func TestExecuteRestoreRoundTripsPublishedExtractionAndManifestMetadata(t *testi
|
|||||||
if lane.Contract == nil || lane.Contract.SchemaID != "encounters" || lane.ExternalProvenance == nil || lane.ExternalProvenance.RunID != "notarius-run-1" {
|
if lane.Contract == nil || lane.Contract.SchemaID != "encounters" || lane.ExternalProvenance == nil || lane.ExternalProvenance.RunID != "notarius-run-1" {
|
||||||
t.Fatalf("restored extraction metadata = %#v", lane)
|
t.Fatalf("restored extraction metadata = %#v", lane)
|
||||||
}
|
}
|
||||||
|
if want := filepath.Join(sessionRoot, "artifacts", "encounters.json"); lane.LocalPath != want {
|
||||||
|
t.Fatalf("rebased extraction path = %q, want %q", lane.LocalPath, want)
|
||||||
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
func TestExecuteRestoreIncludeAudioRestoresAudio(t *testing.T) {
|
func TestExecuteRestoreIncludeAudioRestoresAudio(t *testing.T) {
|
||||||
@@ -200,7 +205,7 @@ func TestExecuteRestoreRestoresPreviousCacheWhenPresent(t *testing.T) {
|
|||||||
if err != nil {
|
if err != nil {
|
||||||
t.Fatalf("read restored previous manifest: %v", err)
|
t.Fatalf("read restored previous manifest: %v", err)
|
||||||
}
|
}
|
||||||
if !strings.Contains(string(previousManifestBytes), `"session_id":"2026-04-26"`) {
|
if !strings.Contains(string(previousManifestBytes), `"session_id": "2026-04-26"`) {
|
||||||
t.Fatalf("restored previous manifest = %q, want previous session id", string(previousManifestBytes))
|
t.Fatalf("restored previous manifest = %q, want previous session id", string(previousManifestBytes))
|
||||||
}
|
}
|
||||||
mustReadEquals(t, filepath.Join(sessionRoot, "previous", "artifacts", "session_recap.md"), "# previous recap\n")
|
mustReadEquals(t, filepath.Join(sessionRoot, "previous", "artifacts", "session_recap.md"), "# previous recap\n")
|
||||||
@@ -293,6 +298,9 @@ func TestExecuteRestoreForceOverwritesDifferingFile(t *testing.T) {
|
|||||||
if !report.Force {
|
if !report.Force {
|
||||||
t.Fatalf("report force = %v, want true", report.Force)
|
t.Fatalf("report force = %v, want true", report.Force)
|
||||||
}
|
}
|
||||||
|
if _, err := os.Stat(artifacts.SessionRestoreMarkerPathForCampaign(workspaceRoot, cfg.Session.Campaign, cfg.Session.SessionID)); !os.IsNotExist(err) {
|
||||||
|
t.Fatalf("restore marker should be cleared after success; stat err=%v", err)
|
||||||
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
func TestExecuteRestoreForceOverwritesDifferingPreviousCacheFile(t *testing.T) {
|
func TestExecuteRestoreForceOverwritesDifferingPreviousCacheFile(t *testing.T) {
|
||||||
@@ -335,14 +343,11 @@ func TestExecuteRestoreLockConflictFailsAndWritesNothing(t *testing.T) {
|
|||||||
|
|
||||||
restoreWithStoreAndRealPhases(t, fake)
|
restoreWithStoreAndRealPhases(t, fake)
|
||||||
|
|
||||||
var stdout bytes.Buffer
|
ctx, cancel := context.WithTimeout(context.Background(), 100*time.Millisecond)
|
||||||
var stderr bytes.Buffer
|
defer cancel()
|
||||||
code := Execute([]string{"session", "restore", "2026-05-03", "--config", pipelinePath, "--campaign-file", campaignPath, "--session", sessionPath}, &stdout, &stderr)
|
err = Restore(ctx, []string{"2026-05-03", "--config", pipelinePath, "--campaign-file", campaignPath, "--session", sessionPath}, &bytes.Buffer{})
|
||||||
if code == 0 {
|
if err == nil || !strings.Contains(err.Error(), "acquire session lock") {
|
||||||
t.Fatal("exit code = 0, want non-zero")
|
t.Fatalf("Restore() error = %v, want lock failure", err)
|
||||||
}
|
|
||||||
if !strings.Contains(stderr.String(), "acquire session lock") {
|
|
||||||
t.Fatalf("stderr = %q, want lock failure", stderr.String())
|
|
||||||
}
|
}
|
||||||
|
|
||||||
sessionRoot := artifacts.SessionWorkDirForCampaign(workspaceRoot, cfg.Session.Campaign, cfg.Session.SessionID)
|
sessionRoot := artifacts.SessionWorkDirForCampaign(workspaceRoot, cfg.Session.Campaign, cfg.Session.SessionID)
|
||||||
@@ -406,6 +411,9 @@ func TestExecuteRestoreInvalidManifestDoesNotCorruptExistingManifest(t *testing.
|
|||||||
if string(afterData) != string(existingData) {
|
if string(afterData) != string(existingData) {
|
||||||
t.Fatalf("local manifest changed after failed restore; before=%q after=%q", string(existingData), string(afterData))
|
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) {
|
func TestExecuteRestorePlanPathMismatchFails(t *testing.T) {
|
||||||
|
|||||||
216
internal/app/restore_manifest_paths.go
Normal file
216
internal/app/restore_manifest_paths.go
Normal file
@@ -0,0 +1,216 @@
|
|||||||
|
package app
|
||||||
|
|
||||||
|
import (
|
||||||
|
"context"
|
||||||
|
"fmt"
|
||||||
|
"io"
|
||||||
|
"path"
|
||||||
|
"path/filepath"
|
||||||
|
"strings"
|
||||||
|
|
||||||
|
"gitea.maximumdirect.net/eric/narratio/internal/config"
|
||||||
|
"gitea.maximumdirect.net/eric/narratio/internal/manifest"
|
||||||
|
)
|
||||||
|
|
||||||
|
func prepareRestoredManifest(
|
||||||
|
ctx context.Context,
|
||||||
|
cfg *config.Config,
|
||||||
|
current *RemoteCurrentState,
|
||||||
|
source io.Reader,
|
||||||
|
destinationRoot string,
|
||||||
|
requireCurrentIdentity bool,
|
||||||
|
) (*manifest.Manifest, error) {
|
||||||
|
manifestStore := &manifest.LocalStore{}
|
||||||
|
m, err := manifestStore.LoadReader(ctx, source)
|
||||||
|
if err != nil {
|
||||||
|
return nil, fmt.Errorf("validate manifest decode: %w", err)
|
||||||
|
}
|
||||||
|
if requireCurrentIdentity {
|
||||||
|
if err := validateRestoredManifestIdentity(cfg, current, m); err != nil {
|
||||||
|
return nil, err
|
||||||
|
}
|
||||||
|
}
|
||||||
|
if err := rebaseRestoredManifestPaths(cfg, m, destinationRoot); err != nil {
|
||||||
|
return nil, err
|
||||||
|
}
|
||||||
|
return m, nil
|
||||||
|
}
|
||||||
|
|
||||||
|
func validateRestoredManifestIdentity(cfg *config.Config, current *RemoteCurrentState, m *manifest.Manifest) error {
|
||||||
|
if cfg == nil || cfg.Session == nil {
|
||||||
|
return fmt.Errorf("resolved session config is required")
|
||||||
|
}
|
||||||
|
requestedSession := strings.TrimSpace(cfg.Session.SessionID)
|
||||||
|
requestedCampaign := strings.TrimSpace(cfg.Session.Campaign)
|
||||||
|
manifestSession := strings.TrimSpace(m.SessionID)
|
||||||
|
manifestCampaign := strings.TrimSpace(m.Campaign)
|
||||||
|
if manifestSession != requestedSession {
|
||||||
|
return fmt.Errorf("manifest session_id %q does not match requested session_id %q", manifestSession, requestedSession)
|
||||||
|
}
|
||||||
|
if manifestCampaign == "" {
|
||||||
|
return fmt.Errorf("manifest campaign is required")
|
||||||
|
}
|
||||||
|
if manifestCampaign != requestedCampaign {
|
||||||
|
return fmt.Errorf("manifest campaign %q does not match requested campaign %q", manifestCampaign, requestedCampaign)
|
||||||
|
}
|
||||||
|
if current != nil {
|
||||||
|
if expected := strings.TrimSpace(current.SessionID); expected != "" && manifestSession != expected {
|
||||||
|
return fmt.Errorf("manifest session_id %q does not match discovered session_id %q", manifestSession, expected)
|
||||||
|
}
|
||||||
|
if expected := strings.TrimSpace(current.Campaign); expected != "" && manifestCampaign != expected {
|
||||||
|
return fmt.Errorf("manifest campaign %q does not match discovered campaign %q", manifestCampaign, expected)
|
||||||
|
}
|
||||||
|
if expected := strings.TrimSpace(current.RunID); expected != "" && strings.TrimSpace(m.RunID) != expected {
|
||||||
|
return fmt.Errorf("manifest run_id %q does not match discovered run_id %q", strings.TrimSpace(m.RunID), expected)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
return nil
|
||||||
|
}
|
||||||
|
|
||||||
|
func rebaseRestoredManifestPaths(cfg *config.Config, m *manifest.Manifest, destinationRoot string) error {
|
||||||
|
if cfg == nil || cfg.Pipeline == nil || m == nil {
|
||||||
|
return fmt.Errorf("resolved config and manifest are required")
|
||||||
|
}
|
||||||
|
destinationRoot = filepath.Clean(strings.TrimSpace(destinationRoot))
|
||||||
|
if destinationRoot == "" || destinationRoot == "." {
|
||||||
|
return fmt.Errorf("restored manifest destination root is required")
|
||||||
|
}
|
||||||
|
producerRoot, hasProducerRoot := restoredManifestSessionRoot(m)
|
||||||
|
rebase := func(field, value string) (string, error) {
|
||||||
|
return rebaseRestoredLocalReference(field, value, producerRoot, hasProducerRoot, destinationRoot)
|
||||||
|
}
|
||||||
|
|
||||||
|
for i := range m.Inputs {
|
||||||
|
value, err := rebase("inputs.path", m.Inputs[i].Path)
|
||||||
|
if err != nil {
|
||||||
|
return err
|
||||||
|
}
|
||||||
|
m.Inputs[i].Path = value
|
||||||
|
// Spool and cache locations are host-local implementation details. They
|
||||||
|
// are deliberately not authoritative after a restore.
|
||||||
|
m.Inputs[i].SpoolPath = ""
|
||||||
|
m.Inputs[i].CachePath = ""
|
||||||
|
}
|
||||||
|
for i := range m.Artifacts {
|
||||||
|
value, err := rebase("artifacts.local_path", m.Artifacts[i].LocalPath)
|
||||||
|
if err != nil {
|
||||||
|
return err
|
||||||
|
}
|
||||||
|
m.Artifacts[i].LocalPath = value
|
||||||
|
}
|
||||||
|
for stageName, record := range m.Stages {
|
||||||
|
if record == nil {
|
||||||
|
continue
|
||||||
|
}
|
||||||
|
for i := range record.Outputs {
|
||||||
|
value, err := rebase("stages."+stageName+".outputs.local_path", record.Outputs[i].LocalPath)
|
||||||
|
if err != nil {
|
||||||
|
return err
|
||||||
|
}
|
||||||
|
record.Outputs[i].LocalPath = value
|
||||||
|
}
|
||||||
|
// Logs and generated configuration files are invocation-local diagnostics,
|
||||||
|
// not restored artifacts. Dropping them prevents a producer-machine path
|
||||||
|
// from becoming a usable local reference.
|
||||||
|
record.Logs = nil
|
||||||
|
record.GeneratedConfigs = nil
|
||||||
|
}
|
||||||
|
|
||||||
|
if runID := strings.TrimSpace(m.RunID); runID != "" {
|
||||||
|
m.LocalWorkDir = filepath.Join(destinationRoot, config.PathRunsDirSegment, runID)
|
||||||
|
} else {
|
||||||
|
m.LocalWorkDir = destinationRoot
|
||||||
|
}
|
||||||
|
m.LocalSpoolDir = ""
|
||||||
|
// A post-publish cleanup record is authority to delete producer-local
|
||||||
|
// directories. It must never cross a restore boundary.
|
||||||
|
m.PostPublishCleanup = nil
|
||||||
|
return nil
|
||||||
|
}
|
||||||
|
|
||||||
|
func restoredManifestSessionRoot(m *manifest.Manifest) (string, bool) {
|
||||||
|
if m == nil {
|
||||||
|
return "", false
|
||||||
|
}
|
||||||
|
runRoot := portableCleanPath(m.LocalWorkDir)
|
||||||
|
runID := strings.TrimSpace(m.RunID)
|
||||||
|
if !portableAbsolutePath(runRoot) || runID == "" || path.Base(runRoot) != runID {
|
||||||
|
return "", false
|
||||||
|
}
|
||||||
|
runsDir := path.Dir(runRoot)
|
||||||
|
if path.Base(runsDir) != config.PathRunsDirSegment {
|
||||||
|
return "", false
|
||||||
|
}
|
||||||
|
return path.Dir(runsDir), true
|
||||||
|
}
|
||||||
|
|
||||||
|
func rebaseRestoredLocalReference(field, value, producerRoot string, hasProducerRoot bool, destinationRoot string) (string, error) {
|
||||||
|
value = strings.TrimSpace(value)
|
||||||
|
if value == "" {
|
||||||
|
return "", nil
|
||||||
|
}
|
||||||
|
if portableDrivePath(value) && !portableAbsolutePath(value) {
|
||||||
|
return "", fmt.Errorf("%s has an unsafe relative path %q", field, value)
|
||||||
|
}
|
||||||
|
if portableAbsolutePath(value) {
|
||||||
|
if !hasProducerRoot {
|
||||||
|
return "", fmt.Errorf("%s has an absolute path without a producer session root", field)
|
||||||
|
}
|
||||||
|
relative, ok := portableRelativeWithinRoot(producerRoot, value)
|
||||||
|
if !ok || relative == "" {
|
||||||
|
return "", fmt.Errorf("%s absolute path is outside the producer session root", field)
|
||||||
|
}
|
||||||
|
resolved, err := joinWithinSessionRoot(destinationRoot, relative)
|
||||||
|
if err != nil {
|
||||||
|
return "", fmt.Errorf("%s: %w", field, err)
|
||||||
|
}
|
||||||
|
return resolved, nil
|
||||||
|
}
|
||||||
|
resolved, err := joinWithinSessionRoot(destinationRoot, strings.ReplaceAll(value, "\\", "/"))
|
||||||
|
if err != nil {
|
||||||
|
return "", fmt.Errorf("%s has an unsafe relative path: %w", field, err)
|
||||||
|
}
|
||||||
|
return resolved, nil
|
||||||
|
}
|
||||||
|
|
||||||
|
func portableRelativeWithinRoot(root, candidate string) (string, bool) {
|
||||||
|
root = strings.TrimSuffix(portableCleanPath(root), "/")
|
||||||
|
candidate = portableCleanPath(candidate)
|
||||||
|
if root == "" || candidate == "" {
|
||||||
|
return "", false
|
||||||
|
}
|
||||||
|
compareRoot, compareCandidate := root, candidate
|
||||||
|
if portableDrivePath(root) || portableDrivePath(candidate) || strings.HasPrefix(root, "//") || strings.HasPrefix(candidate, "//") {
|
||||||
|
compareRoot = strings.ToLower(compareRoot)
|
||||||
|
compareCandidate = strings.ToLower(compareCandidate)
|
||||||
|
}
|
||||||
|
if compareCandidate == compareRoot {
|
||||||
|
return "", true
|
||||||
|
}
|
||||||
|
if !strings.HasPrefix(compareCandidate, compareRoot+"/") {
|
||||||
|
return "", false
|
||||||
|
}
|
||||||
|
return strings.TrimPrefix(candidate, root+"/"), true
|
||||||
|
}
|
||||||
|
|
||||||
|
func portableCleanPath(value string) string {
|
||||||
|
value = strings.ReplaceAll(strings.TrimSpace(value), "\\", "/")
|
||||||
|
if value == "" {
|
||||||
|
return ""
|
||||||
|
}
|
||||||
|
return path.Clean(value)
|
||||||
|
}
|
||||||
|
|
||||||
|
func portableAbsolutePath(value string) bool {
|
||||||
|
value = strings.TrimSpace(value)
|
||||||
|
return strings.HasPrefix(value, "/") || strings.HasPrefix(value, "\\") || (len(value) >= 3 && isASCIIAlpha(value[0]) && value[1] == ':' && (value[2] == '/' || value[2] == '\\'))
|
||||||
|
}
|
||||||
|
|
||||||
|
func portableDrivePath(value string) bool {
|
||||||
|
value = strings.TrimSpace(value)
|
||||||
|
return len(value) >= 2 && isASCIIAlpha(value[0]) && value[1] == ':'
|
||||||
|
}
|
||||||
|
|
||||||
|
func isASCIIAlpha(value byte) bool {
|
||||||
|
return (value >= 'a' && value <= 'z') || (value >= 'A' && value <= 'Z')
|
||||||
|
}
|
||||||
63
internal/app/restore_recovery.go
Normal file
63
internal/app/restore_recovery.go
Normal file
@@ -0,0 +1,63 @@
|
|||||||
|
package app
|
||||||
|
|
||||||
|
import (
|
||||||
|
"encoding/json"
|
||||||
|
"fmt"
|
||||||
|
"os"
|
||||||
|
"strings"
|
||||||
|
"time"
|
||||||
|
|
||||||
|
"gitea.maximumdirect.net/eric/narratio/internal/artifacts"
|
||||||
|
"gitea.maximumdirect.net/eric/narratio/internal/fileops"
|
||||||
|
)
|
||||||
|
|
||||||
|
type restoreMarker struct {
|
||||||
|
StartedAt time.Time `json:"started_at"`
|
||||||
|
}
|
||||||
|
|
||||||
|
func writeRestoreMarker(paths artifacts.SessionPaths) error {
|
||||||
|
marker := restoreMarker{StartedAt: nowUTC()}
|
||||||
|
data, err := json.Marshal(marker)
|
||||||
|
if err != nil {
|
||||||
|
return fmt.Errorf("encode recovery marker: %w", err)
|
||||||
|
}
|
||||||
|
if err := fileops.WriteFileAtomic(paths.RestoreMarkerPath, data, fileops.WorkspaceFileMode); err != nil {
|
||||||
|
return fmt.Errorf("write recovery marker: %w", err)
|
||||||
|
}
|
||||||
|
return nil
|
||||||
|
}
|
||||||
|
|
||||||
|
func clearRestoreMarker(paths artifacts.SessionPaths) error {
|
||||||
|
if strings.TrimSpace(paths.RestoreMarkerPath) == "" {
|
||||||
|
return fmt.Errorf("recovery marker path is required")
|
||||||
|
}
|
||||||
|
if err := os.Remove(paths.RestoreMarkerPath); err != nil && !os.IsNotExist(err) {
|
||||||
|
return fmt.Errorf("remove recovery marker: %w", err)
|
||||||
|
}
|
||||||
|
return nil
|
||||||
|
}
|
||||||
|
|
||||||
|
func requireCompleteRestore(paths artifacts.SessionPaths) error {
|
||||||
|
if strings.TrimSpace(paths.RestoreMarkerPath) == "" {
|
||||||
|
return fmt.Errorf("restore recovery marker path is required")
|
||||||
|
}
|
||||||
|
info, err := os.Lstat(paths.RestoreMarkerPath)
|
||||||
|
if os.IsNotExist(err) {
|
||||||
|
return nil
|
||||||
|
}
|
||||||
|
if err != nil {
|
||||||
|
return fmt.Errorf("check restore recovery marker: %w", err)
|
||||||
|
}
|
||||||
|
if !info.Mode().IsRegular() {
|
||||||
|
return fmt.Errorf("restore recovery marker is not a regular file; rerun session restore")
|
||||||
|
}
|
||||||
|
data, err := os.ReadFile(paths.RestoreMarkerPath)
|
||||||
|
if err != nil {
|
||||||
|
return fmt.Errorf("read restore recovery marker: %w", err)
|
||||||
|
}
|
||||||
|
var marker restoreMarker
|
||||||
|
if err := json.Unmarshal(data, &marker); err != nil || marker.StartedAt.IsZero() {
|
||||||
|
return fmt.Errorf("restore recovery marker is invalid; rerun session restore")
|
||||||
|
}
|
||||||
|
return fmt.Errorf("restore is incomplete; rerun session restore before running pipeline stages")
|
||||||
|
}
|
||||||
187
internal/app/restore_recovery_test.go
Normal file
187
internal/app/restore_recovery_test.go
Normal file
@@ -0,0 +1,187 @@
|
|||||||
|
package app
|
||||||
|
|
||||||
|
import (
|
||||||
|
"bytes"
|
||||||
|
"context"
|
||||||
|
"os"
|
||||||
|
"path/filepath"
|
||||||
|
"strings"
|
||||||
|
"testing"
|
||||||
|
"time"
|
||||||
|
|
||||||
|
"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/manifest"
|
||||||
|
)
|
||||||
|
|
||||||
|
func TestRestoreHoldsTransitionLockBeforePlanning(t *testing.T) {
|
||||||
|
workspaceRoot := t.TempDir()
|
||||||
|
pipelinePath, campaignPath, sessionPath := writeValidConfigFiles(t, workspaceRoot)
|
||||||
|
fake := &storage.FakeBackend{}
|
||||||
|
cfg, _, _, _ := seedRestoreCommittedState(t, fake, pipelinePath, campaignPath, sessionPath)
|
||||||
|
|
||||||
|
origStoreFn := newObjectStoreFromConfigFn
|
||||||
|
origDiscoverFn := discoverRemoteCurrentStateFn
|
||||||
|
origPlanFn := buildRestorePlanFn
|
||||||
|
t.Cleanup(func() {
|
||||||
|
newObjectStoreFromConfigFn = origStoreFn
|
||||||
|
discoverRemoteCurrentStateFn = origDiscoverFn
|
||||||
|
buildRestorePlanFn = origPlanFn
|
||||||
|
})
|
||||||
|
|
||||||
|
current, err := discoverRemoteCurrentState(context.Background(), cfg, fake)
|
||||||
|
if err != nil {
|
||||||
|
t.Fatalf("discoverRemoteCurrentState() error = %v", err)
|
||||||
|
}
|
||||||
|
planning := make(chan struct{})
|
||||||
|
releasePlanning := make(chan struct{})
|
||||||
|
newObjectStoreFromConfigFn = func(context.Context, *config.Config) (storage.ObjectStore, error) { return fake, nil }
|
||||||
|
discoverRemoteCurrentStateFn = func(context.Context, *config.Config, storage.ObjectStore) (*RemoteCurrentState, error) {
|
||||||
|
return current, nil
|
||||||
|
}
|
||||||
|
buildRestorePlanFn = func(context.Context, *config.Config, *RemoteCurrentState, storage.ObjectStore, RestorePlanOptions) (*RestorePlan, error) {
|
||||||
|
close(planning)
|
||||||
|
<-releasePlanning
|
||||||
|
return &RestorePlan{}, nil
|
||||||
|
}
|
||||||
|
|
||||||
|
restoreDone := make(chan error, 1)
|
||||||
|
go func() {
|
||||||
|
restoreDone <- Restore(context.Background(), []string{
|
||||||
|
cfg.Session.SessionID,
|
||||||
|
"--config", pipelinePath,
|
||||||
|
"--campaign-file", campaignPath,
|
||||||
|
"--session", sessionPath,
|
||||||
|
}, &bytes.Buffer{})
|
||||||
|
}()
|
||||||
|
<-planning
|
||||||
|
|
||||||
|
runnerCtx, cancelRunner := context.WithTimeout(context.Background(), 100*time.Millisecond)
|
||||||
|
defer cancelRunner()
|
||||||
|
if _, err := executeStages(runnerCtx, cfg, nil, RunOptions{}); err == nil || !strings.Contains(err.Error(), "acquire session lock") {
|
||||||
|
t.Fatalf("runner while restore plans error = %v, want session-lock failure", err)
|
||||||
|
}
|
||||||
|
|
||||||
|
close(releasePlanning)
|
||||||
|
if err := <-restoreDone; err != nil {
|
||||||
|
t.Fatalf("Restore() error = %v", err)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
func TestExecuteStagesRefusesIncompleteRestore(t *testing.T) {
|
||||||
|
cfg := testConfig(t)
|
||||||
|
paths := artifacts.NewLocalStore(cfg.Pipeline.Workspace.Root).SessionPathsFor(cfg.Session.Campaign, cfg.Session.SessionID)
|
||||||
|
if err := writeRestoreMarker(paths); err != nil {
|
||||||
|
t.Fatalf("writeRestoreMarker() error = %v", err)
|
||||||
|
}
|
||||||
|
|
||||||
|
_, err := executeStages(context.Background(), cfg, nil, RunOptions{})
|
||||||
|
if err == nil || !strings.Contains(err.Error(), "restore is incomplete") {
|
||||||
|
t.Fatalf("executeStages() error = %v, want incomplete restore failure", err)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
func TestForcedRestoreRetryClearsIncompleteMarker(t *testing.T) {
|
||||||
|
workspaceRoot := t.TempDir()
|
||||||
|
pipelinePath, campaignPath, sessionPath := writeValidConfigFiles(t, workspaceRoot)
|
||||||
|
fake := &storage.FakeBackend{}
|
||||||
|
cfg, sessionPrefix, _, _ := seedRestoreCommittedState(t, fake, pipelinePath, campaignPath, sessionPath)
|
||||||
|
artifactKey := sessionPrefix + "artifacts/session_recap.md"
|
||||||
|
transcriptKey := sessionPrefix + "transcripts/full.json"
|
||||||
|
seedRestoreObject(fake, artifactKey, []byte("remote recap\n"))
|
||||||
|
seedRestoreObject(fake, transcriptKey, []byte("remote transcript\n"))
|
||||||
|
|
||||||
|
sessionRoot := artifacts.SessionWorkDirForCampaign(workspaceRoot, cfg.Session.Campaign, cfg.Session.SessionID)
|
||||||
|
mustWriteTestFile(t, filepath.Join(sessionRoot, "artifacts", "session_recap.md"), "local recap\n")
|
||||||
|
mustWriteTestFile(t, filepath.Join(sessionRoot, "transcripts", "full.json"), "local transcript\n")
|
||||||
|
fake.DownloadHook = func(call storage.FakeDownloadCall) error {
|
||||||
|
if call.Key == transcriptKey {
|
||||||
|
return context.Canceled
|
||||||
|
}
|
||||||
|
return nil
|
||||||
|
}
|
||||||
|
restoreWithStoreAndRealPhases(t, fake)
|
||||||
|
|
||||||
|
var stdout bytes.Buffer
|
||||||
|
var stderr bytes.Buffer
|
||||||
|
args := []string{"session", "restore", cfg.Session.SessionID, "--config", pipelinePath, "--campaign-file", campaignPath, "--session", sessionPath, "--force"}
|
||||||
|
if code := Execute(args, &stdout, &stderr); code == 0 {
|
||||||
|
t.Fatal("forced restore exit code = 0, want partial-install failure")
|
||||||
|
}
|
||||||
|
markerPath := artifacts.SessionRestoreMarkerPathForCampaign(workspaceRoot, cfg.Session.Campaign, cfg.Session.SessionID)
|
||||||
|
if _, err := os.Stat(markerPath); err != nil {
|
||||||
|
t.Fatalf("incomplete restore marker missing after partial forced restore: %v", err)
|
||||||
|
}
|
||||||
|
mustReadEquals(t, filepath.Join(sessionRoot, "artifacts", "session_recap.md"), "remote recap\n")
|
||||||
|
|
||||||
|
fake.DownloadHook = nil
|
||||||
|
stdout.Reset()
|
||||||
|
stderr.Reset()
|
||||||
|
if code := Execute(args, &stdout, &stderr); code != 0 {
|
||||||
|
t.Fatalf("restore retry exit code = %d, want 0; stderr=%q", code, stderr.String())
|
||||||
|
}
|
||||||
|
if _, err := os.Stat(markerPath); !os.IsNotExist(err) {
|
||||||
|
t.Fatalf("incomplete restore marker should be cleared after retry; stat err=%v", err)
|
||||||
|
}
|
||||||
|
mustReadEquals(t, filepath.Join(sessionRoot, "transcripts", "full.json"), "remote transcript\n")
|
||||||
|
}
|
||||||
|
|
||||||
|
func TestRebaseRestoredManifestPathsRejectsUnsafeReferencesAndRebasesPortablePaths(t *testing.T) {
|
||||||
|
for _, test := range []struct {
|
||||||
|
name string
|
||||||
|
workDir string
|
||||||
|
output string
|
||||||
|
wantPath string
|
||||||
|
wantError string
|
||||||
|
}{
|
||||||
|
{
|
||||||
|
name: "unix absolute path within producer root",
|
||||||
|
workDir: "/producer/work/sample-campaign/2026-05-03/runs/20260519T010203Z-a1b2c3d4",
|
||||||
|
output: "/producer/work/sample-campaign/2026-05-03/artifacts/recap.md",
|
||||||
|
wantPath: "artifacts/recap.md",
|
||||||
|
},
|
||||||
|
{
|
||||||
|
name: "windows absolute path within producer root",
|
||||||
|
workDir: `C:\producer\work\sample-campaign\2026-05-03\runs\20260519T010203Z-a1b2c3d4`,
|
||||||
|
output: `C:\producer\work\sample-campaign\2026-05-03\artifacts\recap.md`,
|
||||||
|
wantPath: "artifacts/recap.md",
|
||||||
|
},
|
||||||
|
{
|
||||||
|
name: "absolute path outside producer root",
|
||||||
|
workDir: "/producer/work/sample-campaign/2026-05-03/runs/20260519T010203Z-a1b2c3d4",
|
||||||
|
output: "/another-machine/recap.md",
|
||||||
|
wantError: "outside the producer session root",
|
||||||
|
},
|
||||||
|
{
|
||||||
|
name: "relative traversal",
|
||||||
|
workDir: "/producer/work/sample-campaign/2026-05-03/runs/20260519T010203Z-a1b2c3d4",
|
||||||
|
output: "../recap.md",
|
||||||
|
wantError: "unsafe relative path",
|
||||||
|
},
|
||||||
|
} {
|
||||||
|
t.Run(test.name, func(t *testing.T) {
|
||||||
|
cfg := testConfig(t)
|
||||||
|
m := manifest.New(cfg.Session.SessionID, nowUTC())
|
||||||
|
m.Campaign = cfg.Session.Campaign
|
||||||
|
m.RunID = "20260519T010203Z-a1b2c3d4"
|
||||||
|
m.LocalWorkDir = test.workDir
|
||||||
|
m.Stages["analyze"] = &manifest.StageRecord{Outputs: []manifest.ArtifactRecord{{Kind: "recap", LocalPath: test.output}}}
|
||||||
|
destination := artifacts.SessionWorkDirForCampaign(cfg.Pipeline.Workspace.Root, cfg.Session.Campaign, cfg.Session.SessionID)
|
||||||
|
|
||||||
|
err := rebaseRestoredManifestPaths(cfg, m, destination)
|
||||||
|
if test.wantError != "" {
|
||||||
|
if err == nil || !strings.Contains(err.Error(), test.wantError) {
|
||||||
|
t.Fatalf("rebaseRestoredManifestPaths() error = %v, want %q", err, test.wantError)
|
||||||
|
}
|
||||||
|
return
|
||||||
|
}
|
||||||
|
if err != nil {
|
||||||
|
t.Fatalf("rebaseRestoredManifestPaths() error = %v", err)
|
||||||
|
}
|
||||||
|
if got, want := m.Stages["analyze"].Outputs[0].LocalPath, filepath.Join(destination, test.wantPath); got != want {
|
||||||
|
t.Fatalf("rebased local path = %q, want %q", got, want)
|
||||||
|
}
|
||||||
|
})
|
||||||
|
}
|
||||||
|
}
|
||||||
@@ -211,7 +211,7 @@ previous_session_id: 2026-04-26
|
|||||||
if err != nil {
|
if err != nil {
|
||||||
t.Fatalf("read restored previous manifest: %v", err)
|
t.Fatalf("read restored previous manifest: %v", err)
|
||||||
}
|
}
|
||||||
if !strings.Contains(string(previousManifestBytes), `"session_id":"2026-04-26"`) {
|
if !strings.Contains(string(previousManifestBytes), `"session_id": "2026-04-26"`) {
|
||||||
t.Fatalf("restored previous manifest = %q, want previous session id", string(previousManifestBytes))
|
t.Fatalf("restored previous manifest = %q, want previous session id", string(previousManifestBytes))
|
||||||
}
|
}
|
||||||
mustReadEquals(t, filepath.Join(sessionRoot, "previous", "artifacts", "session_recap.md"), "# previous recap\n")
|
mustReadEquals(t, filepath.Join(sessionRoot, "previous", "artifacts", "session_recap.md"), "# previous recap\n")
|
||||||
|
|||||||
@@ -67,13 +67,6 @@ func executeStages(ctx context.Context, cfg *config.Config, stages []stage.Stage
|
|||||||
identity.Campaign,
|
identity.Campaign,
|
||||||
identity.SessionID,
|
identity.SessionID,
|
||||||
)
|
)
|
||||||
if existing, present, err := loadManifestAtPathIfPresent(ctx, env.ManifestStore, manifestPath); err != nil {
|
|
||||||
return nil, err
|
|
||||||
} else if present {
|
|
||||||
if err := identity.validateSessionManifest(existing); err != nil {
|
|
||||||
return nil, err
|
|
||||||
}
|
|
||||||
}
|
|
||||||
if env.ArtifactStore == nil {
|
if env.ArtifactStore == nil {
|
||||||
env.ArtifactStore = artifacts.NewLocalStore(cfg.Pipeline.Workspace.Root)
|
env.ArtifactStore = artifacts.NewLocalStore(cfg.Pipeline.Workspace.Root)
|
||||||
}
|
}
|
||||||
@@ -104,6 +97,9 @@ func executeStages(ctx context.Context, cfg *config.Config, stages []stage.Stage
|
|||||||
if paths.ManifestPath != manifestPath {
|
if paths.ManifestPath != manifestPath {
|
||||||
return nil, fmt.Errorf("prepared manifest path %q does not match resolved manifest path %q", paths.ManifestPath, manifestPath)
|
return nil, fmt.Errorf("prepared manifest path %q does not match resolved manifest path %q", paths.ManifestPath, manifestPath)
|
||||||
}
|
}
|
||||||
|
if err := requireCompleteRestore(paths); err != nil {
|
||||||
|
return nil, err
|
||||||
|
}
|
||||||
m, present, err := loadManifestAtPathIfPresent(ctx, env.ManifestStore, manifestPath)
|
m, present, err := loadManifestAtPathIfPresent(ctx, env.ManifestStore, manifestPath)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return nil, err
|
return nil, err
|
||||||
|
|||||||
@@ -1076,8 +1076,8 @@ func TestExecuteStagesRejectsPersistedManifestIdentityMismatch(t *testing.T) {
|
|||||||
t.Fatal("persisted manifest changed after identity rejection")
|
t.Fatal("persisted manifest changed after identity rejection")
|
||||||
}
|
}
|
||||||
paths := artifacts.NewLocalStore(cfg.Pipeline.Workspace.Root).SessionPathsFor(cfg.Session.Campaign, cfg.Session.SessionID)
|
paths := artifacts.NewLocalStore(cfg.Pipeline.Workspace.Root).SessionPathsFor(cfg.Session.Campaign, cfg.Session.SessionID)
|
||||||
if _, err := os.Stat(paths.LockPath); !os.IsNotExist(err) {
|
if _, err := os.Stat(paths.LockPath); err != nil {
|
||||||
t.Fatalf("lock path stat error = %v, want no lock side effect", err)
|
t.Fatalf("lock path stat error = %v, want transition lock", err)
|
||||||
}
|
}
|
||||||
})
|
})
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -32,6 +32,7 @@ type SessionPaths struct {
|
|||||||
PreviousArtifactsDir string
|
PreviousArtifactsDir string
|
||||||
ManifestPath string
|
ManifestPath string
|
||||||
LockPath string
|
LockPath string
|
||||||
|
RestoreMarkerPath string
|
||||||
}
|
}
|
||||||
|
|
||||||
// SessionWorkDirForCampaign returns the canonical campaign-aware work directory for one session.
|
// SessionWorkDirForCampaign returns the canonical campaign-aware work directory for one session.
|
||||||
@@ -44,6 +45,12 @@ func SessionManifestPathForCampaign(rootDir, campaign, sessionID string) string
|
|||||||
return filepath.Join(SessionWorkDirForCampaign(rootDir, campaign, sessionID), config.PathManifestFile)
|
return filepath.Join(SessionWorkDirForCampaign(rootDir, campaign, sessionID), config.PathManifestFile)
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// SessionRestoreMarkerPathForCampaign returns the durable recovery marker for
|
||||||
|
// a restore that has started changing one session's local state.
|
||||||
|
func SessionRestoreMarkerPathForCampaign(rootDir, campaign, sessionID string) string {
|
||||||
|
return filepath.Join(SessionWorkDirForCampaign(rootDir, campaign, sessionID), ".restore-incomplete.json")
|
||||||
|
}
|
||||||
|
|
||||||
// SessionRunsDirForCampaign returns the canonical runs directory for one session.
|
// SessionRunsDirForCampaign returns the canonical runs directory for one session.
|
||||||
func SessionRunsDirForCampaign(rootDir, campaign, sessionID string) string {
|
func SessionRunsDirForCampaign(rootDir, campaign, sessionID string) string {
|
||||||
return filepath.Join(SessionWorkDirForCampaign(rootDir, campaign, sessionID), config.PathRunsDirSegment)
|
return filepath.Join(SessionWorkDirForCampaign(rootDir, campaign, sessionID), config.PathRunsDirSegment)
|
||||||
@@ -264,5 +271,6 @@ func buildSessionPathsFromRoot(workspaceRoot, campaign, sessionID, root string)
|
|||||||
PreviousArtifactsDir: filepath.Join(root, config.PathPreviousDirSegment, config.PathArtifactsDirSegment),
|
PreviousArtifactsDir: filepath.Join(root, config.PathPreviousDirSegment, config.PathArtifactsDirSegment),
|
||||||
ManifestPath: filepath.Join(root, config.PathManifestFile),
|
ManifestPath: filepath.Join(root, config.PathManifestFile),
|
||||||
LockPath: filepath.Join(root, config.PathLockFile),
|
LockPath: filepath.Join(root, config.PathLockFile),
|
||||||
|
RestoreMarkerPath: filepath.Join(root, ".restore-incomplete.json"),
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|||||||
Reference in New Issue
Block a user