Validate published output declarations
This commit is contained in:
@@ -184,6 +184,11 @@ For a new-protocol current state, the pointer-selected immutable commit is the
|
|||||||
complete restore authority. Callers receive its declared object identities and
|
complete restore authority. Callers receive its declared object identities and
|
||||||
must not supplement them by listing mutable session prefixes.
|
must not supplement them by listing mutable session prefixes.
|
||||||
|
|
||||||
|
A `published_output` declaration must use a canonical publish-compatible source
|
||||||
|
ID and an exact `runs/{run_id}/published/{destination}` key in the same session.
|
||||||
|
Only this declaration type may refer to an earlier run; other declared objects
|
||||||
|
remain confined to the selected commit's run.
|
||||||
|
|
||||||
The reader opens each small control object directly and enforces owner-specific
|
The reader opens each small control object directly and enforces owner-specific
|
||||||
limits before decoding: 64 KiB for the mutable commit pointer, 4 MiB for the
|
limits before decoding: 64 KiB for the mutable commit pointer, 4 MiB for the
|
||||||
immutable commit manifest, and 8 MiB for the selected session manifest. These
|
immutable commit manifest, and 8 MiB for the selected session manifest. These
|
||||||
|
|||||||
@@ -178,7 +178,7 @@ func remotePublishedOutputAvailabilityForCurrent(
|
|||||||
out[publishedOutputRemoteStateKey(source, dest)] = state + "remote=missing"
|
out[publishedOutputRemoteStateKey(source, dest)] = state + "remote=missing"
|
||||||
continue
|
continue
|
||||||
}
|
}
|
||||||
identity, err := artifacts.ParseImmutableRunObjectKey(current.SessionPrefix, declaration.DestinationKey)
|
identity, err := artifacts.ParsePublishedOutputObjectKey(current.SessionPrefix, declaration.DestinationKey)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
out[publishedOutputRemoteStateKey(source, dest)] = state + "remote=error"
|
out[publishedOutputRemoteStateKey(source, dest)] = state + "remote=error"
|
||||||
continue
|
continue
|
||||||
|
|||||||
@@ -855,7 +855,7 @@ func seedOperatorCommittedOutputs(t *testing.T, fake *storage.FakeBackend, outpu
|
|||||||
}
|
}
|
||||||
sort.Strings(sources)
|
sort.Strings(sources)
|
||||||
for _, source := range sources {
|
for _, source := range sources {
|
||||||
key := artifacts.S3RunRelativeDestinationKey(artifacts.S3RunPrefix(sessionPrefix, runID), outputs[source])
|
key := artifacts.S3RunRelativeDestinationKey(artifacts.S3RunPrefix(sessionPrefix, runID), "published/"+outputs[source])
|
||||||
fake.SeedObject(storage.FakeObject{Key: key, Data: []byte(`{"available":true}`)})
|
fake.SeedObject(storage.FakeObject{Key: key, Data: []byte(`{"available":true}`)})
|
||||||
declarations = append(declarations, remoteRestoreArtifact(fake, artifacts.RemoteArtifactTypePublishedOutput, source, key))
|
declarations = append(declarations, remoteRestoreArtifact(fake, artifacts.RemoteArtifactTypePublishedOutput, source, key))
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -50,7 +50,7 @@ func TestCommittedRestoreDownloadsCarriedOutputFromDeclaredRun(t *testing.T) {
|
|||||||
"transcripts/full.json": []byte("carried\n"),
|
"transcripts/full.json": []byte("carried\n"),
|
||||||
})
|
})
|
||||||
declaration := ¤t.Commit.Artifacts[1]
|
declaration := ¤t.Commit.Artifacts[1]
|
||||||
oldKey := artifacts.S3RunRelativeDestinationKey(artifacts.S3RunPrefix(current.SessionPrefix, "20260512T010203Z-11223344"), "transcripts/full.json")
|
oldKey := artifacts.S3RunRelativeDestinationKey(artifacts.S3RunPrefix(current.SessionPrefix, "20260512T010203Z-11223344"), "published/transcripts/full.json")
|
||||||
fake.SeedObject(storage.FakeObject{Key: oldKey, Data: []byte("carried\n")})
|
fake.SeedObject(storage.FakeObject{Key: oldKey, Data: []byte("carried\n")})
|
||||||
oldObject := fake.Objects[oldKey]
|
oldObject := fake.Objects[oldKey]
|
||||||
declaration.DestinationKey = oldKey
|
declaration.DestinationKey = oldKey
|
||||||
@@ -145,7 +145,7 @@ func TestCommittedStatusReportsCarriedOutputOriginAndChecksExactKey(t *testing.T
|
|||||||
"transcripts/full.json": []byte("carried\n"),
|
"transcripts/full.json": []byte("carried\n"),
|
||||||
})
|
})
|
||||||
const originRun = "20260512T010203Z-11223344"
|
const originRun = "20260512T010203Z-11223344"
|
||||||
oldKey := artifacts.S3RunRelativeDestinationKey(artifacts.S3RunPrefix(current.SessionPrefix, originRun), "transcripts/full.json")
|
oldKey := artifacts.S3RunRelativeDestinationKey(artifacts.S3RunPrefix(current.SessionPrefix, originRun), "published/transcripts/full.json")
|
||||||
fake.SeedObject(storage.FakeObject{Key: oldKey, Data: []byte("carried\n")})
|
fake.SeedObject(storage.FakeObject{Key: oldKey, Data: []byte("carried\n")})
|
||||||
current.Commit.Artifacts[1].Source = "narratio.transcript.final_trimmed"
|
current.Commit.Artifacts[1].Source = "narratio.transcript.final_trimmed"
|
||||||
current.Commit.Artifacts[1].DestinationKey = oldKey
|
current.Commit.Artifacts[1].DestinationKey = oldKey
|
||||||
@@ -223,7 +223,7 @@ func TestCommittedRestoreRejectsChangedDeclaredObjectBeforeManifestInstall(t *te
|
|||||||
if err != nil {
|
if err != nil {
|
||||||
t.Fatalf("buildRestorePlan() error = %v", err)
|
t.Fatalf("buildRestorePlan() error = %v", err)
|
||||||
}
|
}
|
||||||
key := artifacts.S3RunRelativeDestinationKey(artifacts.S3RunPrefix(current.SessionPrefix, current.RunID), "transcripts/full.json")
|
key := artifacts.S3RunRelativeDestinationKey(artifacts.S3RunPrefix(current.SessionPrefix, current.RunID), "published/transcripts/full.json")
|
||||||
fake.SeedObject(storage.FakeObject{Key: key, Data: []byte("changed bytes\n")})
|
fake.SeedObject(storage.FakeObject{Key: key, Data: []byte("changed bytes\n")})
|
||||||
|
|
||||||
if _, err := executeRestorePlan(context.Background(), cfg, current, plan, nil, fake); err == nil {
|
if _, err := executeRestorePlan(context.Background(), cfg, current, plan, nil, fake); err == nil {
|
||||||
@@ -245,7 +245,7 @@ func TestCommittedRestoreRejectsChangedDeclaredObjectGeneration(t *testing.T) {
|
|||||||
if err != nil {
|
if err != nil {
|
||||||
t.Fatalf("buildRestorePlan() error = %v", err)
|
t.Fatalf("buildRestorePlan() error = %v", err)
|
||||||
}
|
}
|
||||||
key := artifacts.S3RunRelativeDestinationKey(artifacts.S3RunPrefix(current.SessionPrefix, current.RunID), "transcripts/full.json")
|
key := artifacts.S3RunRelativeDestinationKey(artifacts.S3RunPrefix(current.SessionPrefix, current.RunID), "published/transcripts/full.json")
|
||||||
fake.SeedObject(storage.FakeObject{Key: key, Data: []byte("committed bytes\n"), ETag: "replacement-generation"})
|
fake.SeedObject(storage.FakeObject{Key: key, Data: []byte("committed bytes\n"), ETag: "replacement-generation"})
|
||||||
|
|
||||||
if _, err := executeRestorePlan(context.Background(), cfg, current, plan, nil, fake); err == nil {
|
if _, err := executeRestorePlan(context.Background(), cfg, current, plan, nil, fake); err == nil {
|
||||||
@@ -281,7 +281,7 @@ func TestCommittedRestoreRejectsChangedAudioBeforeInstall(t *testing.T) {
|
|||||||
if err != nil {
|
if err != nil {
|
||||||
t.Fatalf("buildRestorePlan() error = %v", err)
|
t.Fatalf("buildRestorePlan() error = %v", err)
|
||||||
}
|
}
|
||||||
key := artifacts.S3RunRelativeDestinationKey(artifacts.S3RunPrefix(current.SessionPrefix, current.RunID), "audio/alice.flac")
|
key := artifacts.S3RunRelativeDestinationKey(artifacts.S3RunPrefix(current.SessionPrefix, current.RunID), "published/audio/alice.flac")
|
||||||
generation := test.generation
|
generation := test.generation
|
||||||
if generation == "" {
|
if generation == "" {
|
||||||
generation = fake.Objects[key].ETag
|
generation = fake.Objects[key].ETag
|
||||||
@@ -319,7 +319,7 @@ func TestCommittedRestoreRejectsMissingDeclaredObjectBeforeManifestInstall(t *te
|
|||||||
if err != nil {
|
if err != nil {
|
||||||
t.Fatalf("buildRestorePlan() error = %v", err)
|
t.Fatalf("buildRestorePlan() error = %v", err)
|
||||||
}
|
}
|
||||||
missingKey := artifacts.S3RunRelativeDestinationKey(artifacts.S3RunPrefix(current.SessionPrefix, current.RunID), "transcripts/full.json")
|
missingKey := artifacts.S3RunRelativeDestinationKey(artifacts.S3RunPrefix(current.SessionPrefix, current.RunID), "published/transcripts/full.json")
|
||||||
fake.DownloadHook = func(call storage.FakeDownloadCall) error {
|
fake.DownloadHook = func(call storage.FakeDownloadCall) error {
|
||||||
if call.Key == missingKey {
|
if call.Key == missingKey {
|
||||||
return os.ErrNotExist
|
return os.ErrNotExist
|
||||||
@@ -382,9 +382,10 @@ func seedCommittedRestoreSnapshot(t *testing.T, cfg *config.Config, fake *storag
|
|||||||
}
|
}
|
||||||
sort.Strings(paths)
|
sort.Strings(paths)
|
||||||
for _, relative := range paths {
|
for _, relative := range paths {
|
||||||
key := artifacts.S3RunRelativeDestinationKey(artifacts.S3RunPrefix(sessionPrefix, runID), relative)
|
key := artifacts.S3RunRelativeDestinationKey(artifacts.S3RunPrefix(sessionPrefix, runID), "published/"+relative)
|
||||||
fake.SeedObject(storage.FakeObject{Key: key, Data: outputs[relative]})
|
fake.SeedObject(storage.FakeObject{Key: key, Data: outputs[relative]})
|
||||||
artifactsByKey = append(artifactsByKey, remoteRestoreArtifact(fake, artifacts.RemoteArtifactTypePublishedOutput, "narratio.test."+strings.ReplaceAll(relative, "/", "."), key))
|
sourceKey := strings.NewReplacer("/", "_", ".", "_", "-", "_").Replace(relative)
|
||||||
|
artifactsByKey = append(artifactsByKey, remoteRestoreArtifact(fake, artifacts.RemoteArtifactTypePublishedOutput, artifacts.ConfiguredArtifactSourceID(sourceKey), key))
|
||||||
}
|
}
|
||||||
commit := artifacts.RemoteCommitManifest{
|
commit := artifacts.RemoteCommitManifest{
|
||||||
FormatVersion: artifacts.RemoteCommitFormatVersion,
|
FormatVersion: artifacts.RemoteCommitFormatVersion,
|
||||||
|
|||||||
@@ -164,7 +164,7 @@ func TestExecuteRestoreIncludeAudioUsesCacheAfterWorkspaceDeletion(t *testing.T)
|
|||||||
if code != 0 {
|
if code != 0 {
|
||||||
t.Fatalf("first restore exit code = %d, want 0; stderr=%q", code, stderr.String())
|
t.Fatalf("first restore exit code = %d, want 0; stderr=%q", code, stderr.String())
|
||||||
}
|
}
|
||||||
committedAudioKey := artifacts.S3RunRelativeDestinationKey(artifacts.S3RunPrefix(sessionPrefix, "20260519T010203Z-a1b2c3d4"), "audio/alice.flac")
|
committedAudioKey := artifacts.S3RunRelativeDestinationKey(artifacts.S3RunPrefix(sessionPrefix, "20260519T010203Z-a1b2c3d4"), "published/audio/alice.flac")
|
||||||
if got := fakeDownloadCount(fake, committedAudioKey); got != 1 {
|
if got := fakeDownloadCount(fake, committedAudioKey); got != 1 {
|
||||||
t.Fatalf("audio downloads after first restore = %d, want 1", got)
|
t.Fatalf("audio downloads after first restore = %d, want 1", got)
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -182,12 +182,11 @@ func restoreLocalRelativePathForCommittedArtifact(sessionPrefix string, artifact
|
|||||||
if artifact.Type != artifacts.RemoteArtifactTypePublishedOutput {
|
if artifact.Type != artifacts.RemoteArtifactTypePublishedOutput {
|
||||||
return "", false, nil
|
return "", false, nil
|
||||||
}
|
}
|
||||||
identity, err := artifacts.ParseImmutableRunObjectKey(sessionPrefix, artifact.DestinationKey)
|
identity, err := artifacts.ParsePublishedOutputObjectKey(sessionPrefix, artifact.DestinationKey)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return "", false, err
|
return "", false, err
|
||||||
}
|
}
|
||||||
rel := strings.TrimPrefix(identity.RelativePath, "published/")
|
cleanRel, err := pathsafe.NormalizeRelativeDestination(identity.Destination)
|
||||||
cleanRel, err := pathsafe.NormalizeRelativeDestination(rel)
|
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return "", false, fmt.Errorf("committed destination is unsafe: %w", err)
|
return "", false, fmt.Errorf("committed destination is unsafe: %w", err)
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -353,14 +353,15 @@ func TestRestorePlanTraversalUnsafeKeyFails(t *testing.T) {
|
|||||||
func seedCurrentRestoreObject(t *testing.T, store *storage.FakeBackend, current *RemoteCurrentState, key string, data []byte) {
|
func seedCurrentRestoreObject(t *testing.T, store *storage.FakeBackend, current *RemoteCurrentState, key string, data []byte) {
|
||||||
t.Helper()
|
t.Helper()
|
||||||
artifactType := artifacts.RemoteArtifactTypePublishedOutput
|
artifactType := artifacts.RemoteArtifactTypePublishedOutput
|
||||||
source := "narratio.restore." + strings.ReplaceAll(strings.TrimPrefix(key, current.SessionPrefix), "/", ".")
|
sourceKey := strings.NewReplacer("/", "_", ".", "_", "-", "_").Replace(strings.TrimPrefix(key, current.SessionPrefix))
|
||||||
|
source := artifacts.ConfiguredArtifactSourceID(sourceKey)
|
||||||
destination := key
|
destination := key
|
||||||
if key == current.CurrentManifestKey {
|
if key == current.CurrentManifestKey {
|
||||||
artifactType = artifacts.RemoteArtifactTypeSessionManifest
|
artifactType = artifacts.RemoteArtifactTypeSessionManifest
|
||||||
source = "session.manifest"
|
source = "session.manifest"
|
||||||
current.ManifestData = append([]byte(nil), data...)
|
current.ManifestData = append([]byte(nil), data...)
|
||||||
} else if strings.HasPrefix(key, current.SessionPrefix) {
|
} else if strings.HasPrefix(key, current.SessionPrefix) {
|
||||||
destination = artifacts.S3RunPrefix(current.SessionPrefix, current.RunID) + strings.TrimPrefix(key, current.SessionPrefix)
|
destination = artifacts.S3RunPrefix(current.SessionPrefix, current.RunID) + "published/" + strings.TrimPrefix(key, current.SessionPrefix)
|
||||||
}
|
}
|
||||||
seedRestoreObject(store, destination, data)
|
seedRestoreObject(store, destination, data)
|
||||||
current.Commit.Artifacts = append(current.Commit.Artifacts, remoteRestoreArtifact(store, artifactType, source, destination))
|
current.Commit.Artifacts = append(current.Commit.Artifacts, remoteRestoreArtifact(store, artifactType, source, destination))
|
||||||
@@ -397,11 +398,11 @@ func seedRestoreObject(store *storage.FakeBackend, key string, data []byte) {
|
|||||||
destination := key
|
destination := key
|
||||||
if strings.HasPrefix(key, sessionPrefix) && !strings.HasPrefix(key, runPrefix) {
|
if strings.HasPrefix(key, sessionPrefix) && !strings.HasPrefix(key, runPrefix) {
|
||||||
relative := strings.TrimPrefix(key, sessionPrefix)
|
relative := strings.TrimPrefix(key, sessionPrefix)
|
||||||
destination = runPrefix + relative
|
destination = runPrefix + "published/" + relative
|
||||||
store.SeedObject(storage.FakeObject{Key: destination, Data: data})
|
store.SeedObject(storage.FakeObject{Key: destination, Data: data})
|
||||||
}
|
}
|
||||||
typeID := artifacts.RemoteArtifactTypePublishedOutput
|
typeID := artifacts.RemoteArtifactTypePublishedOutput
|
||||||
source := "narratio.test"
|
source := artifacts.ConfiguredArtifactSourceID("test")
|
||||||
if destination == artifacts.S3RunSessionManifestKey(sessionPrefix, pointer.RunID) {
|
if destination == artifacts.S3RunSessionManifestKey(sessionPrefix, pointer.RunID) {
|
||||||
typeID, source = artifacts.RemoteArtifactTypeSessionManifest, "session.manifest"
|
typeID, source = artifacts.RemoteArtifactTypeSessionManifest, "session.manifest"
|
||||||
} else if relative := strings.TrimPrefix(destination, runPrefix); strings.HasPrefix(relative, "artifacts/") || strings.HasPrefix(relative, "published/artifacts/") {
|
} else if relative := strings.TrimPrefix(destination, runPrefix); strings.HasPrefix(relative, "artifacts/") || strings.HasPrefix(relative, "published/artifacts/") {
|
||||||
|
|||||||
@@ -95,7 +95,7 @@ func TestForcedRestoreRetryClearsIncompleteMarker(t *testing.T) {
|
|||||||
sessionRoot := artifacts.SessionWorkDirForCampaign(workspaceRoot, cfg.Session.Campaign, cfg.Session.SessionID)
|
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, "artifacts", "session_recap.md"), "local recap\n")
|
||||||
mustWriteTestFile(t, filepath.Join(sessionRoot, "transcripts", "full.json"), "local transcript\n")
|
mustWriteTestFile(t, filepath.Join(sessionRoot, "transcripts", "full.json"), "local transcript\n")
|
||||||
committedTranscriptKey := artifacts.S3RunRelativeDestinationKey(artifacts.S3RunPrefix(sessionPrefix, "20260519T010203Z-a1b2c3d4"), "transcripts/full.json")
|
committedTranscriptKey := artifacts.S3RunRelativeDestinationKey(artifacts.S3RunPrefix(sessionPrefix, "20260519T010203Z-a1b2c3d4"), "published/transcripts/full.json")
|
||||||
fake.DownloadHook = func(call storage.FakeDownloadCall) error {
|
fake.DownloadHook = func(call storage.FakeDownloadCall) error {
|
||||||
if call.Key == committedTranscriptKey {
|
if call.Key == committedTranscriptKey {
|
||||||
return context.Canceled
|
return context.Canceled
|
||||||
|
|||||||
@@ -117,18 +117,14 @@ func (m RemoteCommitManifest) ValidateForSessionPrefix(sessionPrefix string) err
|
|||||||
publishedDestinations := make(map[string]struct{})
|
publishedDestinations := make(map[string]struct{})
|
||||||
for _, artifact := range m.Artifacts {
|
for _, artifact := range m.Artifacts {
|
||||||
if artifact.Type == RemoteArtifactTypePublishedOutput {
|
if artifact.Type == RemoteArtifactTypePublishedOutput {
|
||||||
identity, err := ParseImmutableRunObjectKey(sessionPrefix, artifact.DestinationKey)
|
identity, err := ParsePublishedOutputObjectKey(sessionPrefix, artifact.DestinationKey)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return fmt.Errorf("remote commit published output destination %q: %w", artifact.DestinationKey, err)
|
return fmt.Errorf("remote commit published output destination %q: %w", artifact.DestinationKey, err)
|
||||||
}
|
}
|
||||||
effectiveDestination := strings.TrimPrefix(identity.RelativePath, "published/")
|
if _, duplicate := publishedDestinations[identity.Destination]; duplicate {
|
||||||
if effectiveDestination == "" {
|
return fmt.Errorf("remote commit declares duplicate published output destination %q", identity.Destination)
|
||||||
return fmt.Errorf("remote commit published output destination %q is empty", artifact.DestinationKey)
|
|
||||||
}
|
}
|
||||||
if _, duplicate := publishedDestinations[effectiveDestination]; duplicate {
|
publishedDestinations[identity.Destination] = struct{}{}
|
||||||
return fmt.Errorf("remote commit declares duplicate published output destination %q", effectiveDestination)
|
|
||||||
}
|
|
||||||
publishedDestinations[effectiveDestination] = struct{}{}
|
|
||||||
continue
|
continue
|
||||||
}
|
}
|
||||||
if !strings.HasPrefix(artifact.DestinationKey, runPrefix) {
|
if !strings.HasPrefix(artifact.DestinationKey, runPrefix) {
|
||||||
@@ -163,6 +159,11 @@ func validateRemoteArtifactIdentity(a RemoteArtifact) error {
|
|||||||
if strings.TrimSpace(a.Source) == "" {
|
if strings.TrimSpace(a.Source) == "" {
|
||||||
return fmt.Errorf("source is required")
|
return fmt.Errorf("source is required")
|
||||||
}
|
}
|
||||||
|
if a.Type == RemoteArtifactTypePublishedOutput {
|
||||||
|
if err := validateCanonicalPublishSource(a.Source); err != nil {
|
||||||
|
return fmt.Errorf("published output %w", err)
|
||||||
|
}
|
||||||
|
}
|
||||||
if err := validateRemoteObjectKey(a.DestinationKey); err != nil {
|
if err := validateRemoteObjectKey(a.DestinationKey); err != nil {
|
||||||
return fmt.Errorf("destination key: %w", err)
|
return fmt.Errorf("destination key: %w", err)
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -77,6 +77,24 @@ func TestParseImmutableRunObjectKeyRejectsUnsafeOrMutableKeys(t *testing.T) {
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
func TestParsePublishedOutputObjectKeyRequiresPublishedNamespace(t *testing.T) {
|
||||||
|
prefix := testCurrentSessionPrefix()
|
||||||
|
runID := "20260519T010203Z-a1b2c3d4"
|
||||||
|
key := S3RunRelativeDestinationKey(S3RunPrefix(prefix, runID), "published/artifacts/recap.md")
|
||||||
|
identity, err := ParsePublishedOutputObjectKey(prefix, key)
|
||||||
|
if err != nil {
|
||||||
|
t.Fatalf("ParsePublishedOutputObjectKey() error = %v", err)
|
||||||
|
}
|
||||||
|
if identity.RunID != runID || identity.Destination != "artifacts/recap.md" {
|
||||||
|
t.Fatalf("ParsePublishedOutputObjectKey() = %#v", identity)
|
||||||
|
}
|
||||||
|
|
||||||
|
outside := S3RunRelativeDestinationKey(S3RunPrefix(prefix, runID), "artifacts/recap.md")
|
||||||
|
if _, err := ParsePublishedOutputObjectKey(prefix, outside); err == nil {
|
||||||
|
t.Fatal("ParsePublishedOutputObjectKey() outside published namespace error = nil")
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
func TestRemoteCommitAllowsPublishedOutputFromEarlierSameSessionRunOnly(t *testing.T) {
|
func TestRemoteCommitAllowsPublishedOutputFromEarlierSameSessionRunOnly(t *testing.T) {
|
||||||
commit := testRemoteCommit(t, "20260519T010203Z-a1b2c3d4")
|
commit := testRemoteCommit(t, "20260519T010203Z-a1b2c3d4")
|
||||||
prefix := testCurrentSessionPrefix()
|
prefix := testCurrentSessionPrefix()
|
||||||
@@ -99,6 +117,18 @@ func TestRemoteCommitAllowsPublishedOutputFromEarlierSameSessionRunOnly(t *testi
|
|||||||
if err := commit.ValidateForSessionPrefix(prefix); err == nil {
|
if err := commit.ValidateForSessionPrefix(prefix); err == nil {
|
||||||
t.Fatal("ValidateForSessionPrefix() earlier non-published object error = nil")
|
t.Fatal("ValidateForSessionPrefix() earlier non-published object error = nil")
|
||||||
}
|
}
|
||||||
|
|
||||||
|
commit.Artifacts[1] = published
|
||||||
|
commit.Artifacts[1].DestinationKey = S3RunRelativeDestinationKey(S3RunPrefix(prefix, commit.RunID), "artifacts/recap.md")
|
||||||
|
if err := commit.ValidateForSessionPrefix(prefix); err == nil {
|
||||||
|
t.Fatal("ValidateForSessionPrefix() published output outside published namespace error = nil")
|
||||||
|
}
|
||||||
|
|
||||||
|
commit.Artifacts[1] = published
|
||||||
|
commit.Artifacts[1].Source = "run:artifacts/recap.md"
|
||||||
|
if err := commit.ValidateForSessionPrefix(prefix); err == nil {
|
||||||
|
t.Fatal("ValidateForSessionPrefix() invalid published source error = nil")
|
||||||
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
func TestRemoteCommitRejectsEffectivePublishedDestinationCollisionAcrossRuns(t *testing.T) {
|
func TestRemoteCommitRejectsEffectivePublishedDestinationCollisionAcrossRuns(t *testing.T) {
|
||||||
|
|||||||
@@ -15,7 +15,6 @@ import (
|
|||||||
"time"
|
"time"
|
||||||
|
|
||||||
"gitea.maximumdirect.net/eric/narratio/internal/adapters/storage"
|
"gitea.maximumdirect.net/eric/narratio/internal/adapters/storage"
|
||||||
"gitea.maximumdirect.net/eric/narratio/internal/artifactpolicy"
|
|
||||||
)
|
)
|
||||||
|
|
||||||
const (
|
const (
|
||||||
@@ -66,7 +65,7 @@ func (l RemoteLock) Validate() error {
|
|||||||
if l.FormatVersion != RemoteLockFormatVersion {
|
if l.FormatVersion != RemoteLockFormatVersion {
|
||||||
return fmt.Errorf("unsupported remote lock format version %d", l.FormatVersion)
|
return fmt.Errorf("unsupported remote lock format version %d", l.FormatVersion)
|
||||||
}
|
}
|
||||||
if err := validateRemoteLockSource(l.Source); err != nil {
|
if err := validateCanonicalPublishSource(l.Source); err != nil {
|
||||||
return fmt.Errorf("remote lock source: %w", err)
|
return fmt.Errorf("remote lock source: %w", err)
|
||||||
}
|
}
|
||||||
if strings.TrimSpace(l.Reason) != l.Reason {
|
if strings.TrimSpace(l.Reason) != l.Reason {
|
||||||
@@ -137,7 +136,7 @@ func RemoteLockKey(sessionPrefix, source string) (string, error) {
|
|||||||
if err != nil {
|
if err != nil {
|
||||||
return "", err
|
return "", err
|
||||||
}
|
}
|
||||||
if err := validateRemoteLockSource(source); err != nil {
|
if err := validateCanonicalPublishSource(source); err != nil {
|
||||||
return "", err
|
return "", err
|
||||||
}
|
}
|
||||||
return locksPrefix + remoteLockDigest(source) + ".lock", nil
|
return locksPrefix + remoteLockDigest(source) + ".lock", nil
|
||||||
@@ -267,17 +266,6 @@ func lookupRemoteLockAtKey(ctx context.Context, store storage.ObjectStore, key,
|
|||||||
return RemoteLockLookup{Lock: lock, Present: true, Generation: info.ETag}, nil
|
return RemoteLockLookup{Lock: lock, Present: true, Generation: info.ETag}, nil
|
||||||
}
|
}
|
||||||
|
|
||||||
func validateRemoteLockSource(source string) error {
|
|
||||||
classified, err := artifactpolicy.ClassifySource(source)
|
|
||||||
if err != nil || classified.Kind == artifactpolicy.SourceKindPreviousArtifact {
|
|
||||||
return fmt.Errorf("source %q is not a supported publish source", source)
|
|
||||||
}
|
|
||||||
if classified.ID != source {
|
|
||||||
return fmt.Errorf("source %q is not canonical", source)
|
|
||||||
}
|
|
||||||
return nil
|
|
||||||
}
|
|
||||||
|
|
||||||
func remoteLocksPrefix(sessionPrefix string) (string, error) {
|
func remoteLocksPrefix(sessionPrefix string) (string, error) {
|
||||||
if err := validateRemoteLockSessionPrefix(sessionPrefix); err != nil {
|
if err := validateRemoteLockSessionPrefix(sessionPrefix); err != nil {
|
||||||
return "", err
|
return "", err
|
||||||
|
|||||||
@@ -4,6 +4,7 @@ import (
|
|||||||
"fmt"
|
"fmt"
|
||||||
"strings"
|
"strings"
|
||||||
|
|
||||||
|
"gitea.maximumdirect.net/eric/narratio/internal/artifactpolicy"
|
||||||
"gitea.maximumdirect.net/eric/narratio/internal/config"
|
"gitea.maximumdirect.net/eric/narratio/internal/config"
|
||||||
)
|
)
|
||||||
|
|
||||||
@@ -14,6 +15,13 @@ type ImmutableRunObjectKey struct {
|
|||||||
RelativePath string
|
RelativePath string
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// PublishedOutputObjectKey identifies a published destination beneath one
|
||||||
|
// immutable run in a session.
|
||||||
|
type PublishedOutputObjectKey struct {
|
||||||
|
RunID string
|
||||||
|
Destination string
|
||||||
|
}
|
||||||
|
|
||||||
// ParseImmutableRunObjectKey proves that key belongs to one immutable run in
|
// ParseImmutableRunObjectKey proves that key belongs to one immutable run in
|
||||||
// the expected campaign/session namespace and returns its run and relative path.
|
// the expected campaign/session namespace and returns its run and relative path.
|
||||||
func ParseImmutableRunObjectKey(sessionPrefix, key string) (ImmutableRunObjectKey, error) {
|
func ParseImmutableRunObjectKey(sessionPrefix, key string) (ImmutableRunObjectKey, error) {
|
||||||
@@ -41,6 +49,34 @@ func ParseImmutableRunObjectKey(sessionPrefix, key string) (ImmutableRunObjectKe
|
|||||||
return ImmutableRunObjectKey{RunID: runID, RelativePath: relativePath}, nil
|
return ImmutableRunObjectKey{RunID: runID, RelativePath: relativePath}, nil
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// ParsePublishedOutputObjectKey proves that key is in the published-output
|
||||||
|
// namespace of one immutable run in the expected session.
|
||||||
|
func ParsePublishedOutputObjectKey(sessionPrefix, key string) (PublishedOutputObjectKey, error) {
|
||||||
|
identity, err := ParseImmutableRunObjectKey(sessionPrefix, key)
|
||||||
|
if err != nil {
|
||||||
|
return PublishedOutputObjectKey{}, err
|
||||||
|
}
|
||||||
|
destination, ok := strings.CutPrefix(identity.RelativePath, "published/")
|
||||||
|
if !ok || destination == "" {
|
||||||
|
return PublishedOutputObjectKey{}, fmt.Errorf("immutable object key %q is outside its immutable run published-output prefix", key)
|
||||||
|
}
|
||||||
|
if err := validateRemoteObjectKey(destination); err != nil {
|
||||||
|
return PublishedOutputObjectKey{}, fmt.Errorf("immutable object key %q has invalid published destination: %w", key, err)
|
||||||
|
}
|
||||||
|
return PublishedOutputObjectKey{RunID: identity.RunID, Destination: destination}, nil
|
||||||
|
}
|
||||||
|
|
||||||
|
func validateCanonicalPublishSource(source string) error {
|
||||||
|
classified, err := artifactpolicy.ClassifySource(source)
|
||||||
|
if err != nil || classified.Kind == artifactpolicy.SourceKindPreviousArtifact {
|
||||||
|
return fmt.Errorf("source %q is not a supported publish source", source)
|
||||||
|
}
|
||||||
|
if classified.ID != source {
|
||||||
|
return fmt.Errorf("source %q is not canonical", source)
|
||||||
|
}
|
||||||
|
return nil
|
||||||
|
}
|
||||||
|
|
||||||
func validateRemoteSessionPrefix(sessionPrefix string) error {
|
func validateRemoteSessionPrefix(sessionPrefix string) error {
|
||||||
if sessionPrefix == "" || strings.TrimSpace(sessionPrefix) != sessionPrefix || strings.Contains(sessionPrefix, `\`) || !strings.HasSuffix(sessionPrefix, "/") {
|
if sessionPrefix == "" || strings.TrimSpace(sessionPrefix) != sessionPrefix || strings.Contains(sessionPrefix, `\`) || !strings.HasSuffix(sessionPrefix, "/") {
|
||||||
return fmt.Errorf("session prefix %q must be a canonical relative object prefix", sessionPrefix)
|
return fmt.Errorf("session prefix %q must be a canonical relative object prefix", sessionPrefix)
|
||||||
|
|||||||
@@ -210,15 +210,11 @@ func committedPublishedArtifactBySource(commit *artifacts.RemoteCommitManifest,
|
|||||||
}
|
}
|
||||||
|
|
||||||
func publishedOutputRelativePath(sessionPrefix, destinationKey string) (string, error) {
|
func publishedOutputRelativePath(sessionPrefix, destinationKey string) (string, error) {
|
||||||
identity, err := artifacts.ParseImmutableRunObjectKey(sessionPrefix, destinationKey)
|
identity, err := artifacts.ParsePublishedOutputObjectKey(sessionPrefix, destinationKey)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return "", err
|
return "", err
|
||||||
}
|
}
|
||||||
rel, ok := strings.CutPrefix(identity.RelativePath, "published/")
|
normalized, err := pathsafe.NormalizeRelativeDestination(identity.Destination)
|
||||||
if !ok || strings.TrimSpace(rel) == "" {
|
|
||||||
return "", fmt.Errorf("destination %q is outside its immutable run published-output prefix", destinationKey)
|
|
||||||
}
|
|
||||||
normalized, err := pathsafe.NormalizeRelativeDestination(rel)
|
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return "", fmt.Errorf("published output destination %q: %w", destinationKey, err)
|
return "", fmt.Errorf("published output destination %q: %w", destinationKey, err)
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -90,12 +90,12 @@ func buildPublishOutputPlan(
|
|||||||
}
|
}
|
||||||
if _, locked := lockSet[rule.Source]; locked {
|
if _, locked := lockSet[rule.Source]; locked {
|
||||||
if declaration, found := priorBySource[rule.Source]; found {
|
if declaration, found := priorBySource[rule.Source]; found {
|
||||||
identity, err := artifacts.ParseImmutableRunObjectKey(sessionPrefix, declaration.DestinationKey)
|
identity, err := artifacts.ParsePublishedOutputObjectKey(sessionPrefix, declaration.DestinationKey)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return nil, fmt.Errorf("carried publish output %q: %w", rule.Source, err)
|
return nil, fmt.Errorf("carried publish output %q: %w", rule.Source, err)
|
||||||
}
|
}
|
||||||
outcome.Status = publishOutputCarriedLocked
|
outcome.Status = publishOutputCarriedLocked
|
||||||
outcome.Destination = strings.TrimPrefix(identity.RelativePath, "published/")
|
outcome.Destination = identity.Destination
|
||||||
outcome.DeclaredKey = declaration.DestinationKey
|
outcome.DeclaredKey = declaration.DestinationKey
|
||||||
outcome.OriginRun = identity.RunID
|
outcome.OriginRun = identity.RunID
|
||||||
outcome.Declaration = declaration
|
outcome.Declaration = declaration
|
||||||
@@ -115,7 +115,7 @@ func buildPublishOutputPlan(
|
|||||||
continue
|
continue
|
||||||
}
|
}
|
||||||
declaredKey := publishOutputRemoteKey(runPrefix, rule.Destination)
|
declaredKey := publishOutputRemoteKey(runPrefix, rule.Destination)
|
||||||
identity, err := artifacts.ParseImmutableRunObjectKey(sessionPrefix, declaredKey)
|
identity, err := artifacts.ParsePublishedOutputObjectKey(sessionPrefix, declaredKey)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return nil, fmt.Errorf("publish output %q destination: %w", rule.Source, err)
|
return nil, fmt.Errorf("publish output %q destination: %w", rule.Source, err)
|
||||||
}
|
}
|
||||||
|
|||||||
Reference in New Issue
Block a user