Repository navigation
fix(cli): persist SHA-256 when recording artifacts #236
New issue
Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.
By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.
Already on GitHub? Sign in to your account
Changes from all commits
17de7d4
ad4d685
c01fcf5
73c0e27
966c47a
f0c9275
93407ac
efeb4df
f498450
1b32005
f2d12dc
c6f8223
File filter
Filter by extension
Conversations
Jump to
Diff view
Diff view
There are no files selected for viewing
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,143 @@ | ||
| package library | ||
|
|
||
| import ( | ||
| "context" | ||
| "errors" | ||
| "os" | ||
| "path/filepath" | ||
| "strings" | ||
| "testing" | ||
| "time" | ||
|
|
||
| "github.com/google/uuid" | ||
|
|
||
| "github.com/rabesss/impartus-cli/internal/artifact" | ||
| ) | ||
|
|
||
| func TestDigestManifestFilesRejectsSizeDrift(t *testing.T) { | ||
| path := filepath.Join(t.TempDir(), "lecture.mp4") | ||
| original := []byte{0, 0, 0, 24, 'f', 't', 'y', 'p', 'i', 's', 'o', 'm'} | ||
| if err := os.WriteFile(path, original, 0o600); err != nil { | ||
| t.Fatal(err) | ||
| } | ||
| manifest, err := (ExpectedArtifact{ | ||
| Lecture: artifact.Lecture{TTID: 4, InstituteID: 1, SubjectID: 2, SessionID: 3}, | ||
| Selection: artifact.Selection{Views: "left", Quality: "720"}, | ||
| Files: []ExpectedFile{{Path: path, Role: "video", View: "left", Container: "mp4"}}, | ||
| ProducedAt: time.Date(2026, time.August, 9, 8, 0, 0, 0, time.UTC), | ||
| Producer: artifact.Producer{Name: "impartus", Version: "test"}, | ||
| }).buildManifest() | ||
| if err != nil { | ||
| t.Fatalf("buildManifest() error = %v", err) | ||
| } | ||
|
|
||
| grown := append(append([]byte(nil), original...), 0) | ||
| if err := os.WriteFile(path, grown, 0o600); err != nil { | ||
| t.Fatal(err) | ||
| } | ||
| if _, err := digestManifestFiles(manifest); err == nil { | ||
| t.Fatal("digestManifestFiles() error = nil, want size drift rejection") | ||
| } else if !strings.Contains(err.Error(), "size changed") { | ||
| t.Fatalf("digestManifestFiles() error = %v, want size drift rejection", err) | ||
| } | ||
| } | ||
|
|
||
| func TestDigestManifestFilesKeepsCallerPinnedDigest(t *testing.T) { | ||
| path := filepath.Join(t.TempDir(), "lecture.mp4") | ||
| media := []byte{0, 0, 0, 24, 'f', 't', 'y', 'p', 'i', 's', 'o', 'm'} | ||
| if err := os.WriteFile(path, media, 0o600); err != nil { | ||
| t.Fatal(err) | ||
| } | ||
| const pinnedSHA256 = "c43403fe022af967a0b859d3e14ea12d6633f4c8ad475816b0c55d85896e8e35" | ||
| manifest, err := (ExpectedArtifact{ | ||
| Lecture: artifact.Lecture{TTID: 4, InstituteID: 1, SubjectID: 2, SessionID: 3}, | ||
| Selection: artifact.Selection{Views: "left", Quality: "720"}, | ||
| Files: []ExpectedFile{{Path: path, Role: "video", View: "left", Container: "mp4", SHA256: pinnedSHA256}}, | ||
| ProducedAt: time.Date(2026, time.August, 9, 8, 0, 0, 0, time.UTC), | ||
| Producer: artifact.Producer{Name: "impartus", Version: "test"}, | ||
| }).buildManifest() | ||
| if err != nil { | ||
| t.Fatalf("buildManifest() error = %v", err) | ||
| } | ||
|
|
||
| // A re-hash would invoke the stability proof; fail if the pin is re-read. | ||
| oldValidate := validateStableArtifactFile | ||
| validateStableArtifactFile = func(string, *os.File, os.FileInfo) error { | ||
| return errors.New("pinned digest must pass through without a re-hash") | ||
| } | ||
| t.Cleanup(func() { validateStableArtifactFile = oldValidate }) | ||
|
|
||
| digested, err := digestManifestFiles(manifest) | ||
| if err != nil { | ||
| t.Fatalf("digestManifestFiles() error = %v", err) | ||
| } | ||
| if len(digested.Files) != 1 || digested.Files[0].SHA256 != pinnedSHA256 { | ||
|
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. WARNING: This test cannot detect removal of the pinned-digest skip
Reply with |
||
| t.Fatalf("digestManifestFiles() files = %+v, want pinned %q", digested.Files, pinnedSHA256) | ||
| } | ||
| } | ||
|
|
||
| func TestRecoverInterruptedJobsSkipsJobWhoseOutputsCannotBeDigested(t *testing.T) { | ||
| parent := filepath.Join(t.TempDir(), "state") | ||
| if err := os.Mkdir(parent, 0o700); err != nil { | ||
| t.Fatal(err) | ||
| } | ||
| store, err := Open(context.Background(), Options{Path: filepath.Join(parent, "library.db")}) | ||
| if err != nil { | ||
| t.Fatal(err) | ||
| } | ||
| t.Cleanup(func() { | ||
| if closeErr := store.Close(); closeErr != nil { | ||
| t.Errorf("Close() error = %v", closeErr) | ||
| } | ||
| }) | ||
|
|
||
| ctx := context.Background() | ||
| media := []byte{0, 0, 0, 24, 'f', 't', 'y', 'p', 'i', 's', 'o', 'm'} | ||
| unstablePath := filepath.Join(t.TempDir(), "unstable.mp4") | ||
| stablePath := filepath.Join(t.TempDir(), "stable.mp4") | ||
| for _, path := range []string{unstablePath, stablePath} { | ||
| if writeErr := os.WriteFile(path, media, 0o600); writeErr != nil { | ||
| t.Fatal(writeErr) | ||
| } | ||
| } | ||
| newJob := func(id, path string, ttid int) { | ||
| t.Helper() | ||
| expected := ExpectedArtifact{ | ||
| Lecture: artifact.Lecture{TTID: ttid, InstituteID: 1, SubjectID: 2, SessionID: 3}, | ||
| Selection: artifact.Selection{Views: "left", Quality: "720"}, | ||
| Files: []ExpectedFile{{Path: path, Role: "video", View: "left", Container: "mp4"}}, | ||
| ProducedAt: time.Date(2026, time.August, 9, 10, 0, 0, 0, time.UTC), | ||
| Producer: artifact.Producer{Name: "impartus", Version: "test"}, | ||
| } | ||
| if createErr := store.CreateJob(ctx, JobSpec{ID: id, Kind: JobKindDownload, Expected: expected}); createErr != nil { | ||
| t.Fatalf("CreateJob(%s) error = %v", id, createErr) | ||
| } | ||
| if startErr := store.StartJob(ctx, id); startErr != nil { | ||
| t.Fatalf("StartJob(%s) error = %v", id, startErr) | ||
| } | ||
| } | ||
| unstableJobID := uuid.NewString() | ||
| stableJobID := uuid.NewString() | ||
| newJob(unstableJobID, unstablePath, 90) | ||
| newJob(stableJobID, stablePath, 91) | ||
|
|
||
| oldValidate := validateStableArtifactFile | ||
| validateStableArtifactFile = func(path string, file *os.File, initial os.FileInfo) error { | ||
| if path == unstablePath { | ||
| return errors.New("injected instability") | ||
| } | ||
| return artifact.ValidateStableCompletedFile(path, file, initial) | ||
| } | ||
| t.Cleanup(func() { validateStableArtifactFile = oldValidate }) | ||
|
|
||
| result, err := store.RecoverInterruptedJobs(ctx, JobKindDownload) | ||
| if err != nil { | ||
| t.Fatalf("RecoverInterruptedJobs() error = %v, want per-job containment of digest failures", err) | ||
| } | ||
| if len(result.Pending) != 1 || result.Pending[0] != unstableJobID { | ||
| t.Fatalf("Pending = %+v, want only the unstable job", result.Pending) | ||
| } | ||
| if len(result.Recovered) != 1 || result.Recovered[0] != stableJobID { | ||
| t.Fatalf("Recovered = %+v, want the stable job to recover past the unstable job", result.Recovered) | ||
| } | ||
| } | ||
| Original file line number | Diff line number | Diff line change |
|---|---|---|
|
|
@@ -50,6 +50,9 @@ var ( | |
| ErrJobTerminal = errors.New("library job is already terminal") | ||
| // ErrJobTransition reports a lifecycle operation invalid for the current state. | ||
| ErrJobTransition = errors.New("invalid library job transition") | ||
| // ErrJobOutputUnstable reports completed outputs that changed or became | ||
| // unreadable between validation and the record-time digest. | ||
| ErrJobOutputUnstable = errors.New("library job output changed before recording") | ||
| ) | ||
|
|
||
| // ExpectedFile records a final output before media creation starts. Partial | ||
|
|
@@ -225,9 +228,14 @@ func (store *Store) CompleteJob(ctx context.Context, jobID string, manifest arti | |
|
|
||
| // completeValidatedJob atomically records a manifest that was built and | ||
| // validated against the current file descriptors by the immediate caller. | ||
| // Keeping this separate prevents startup recovery from reopening the same | ||
| // paths between validation and the durable transaction. | ||
| // Empty digests are filled from a stability-checked read before the write | ||
| // transaction begins, so the full-file read never holds the library write | ||
| // lock; a job that turns out to be already completed pays that read anyway. | ||
| func (store *Store) completeValidatedJob(ctx context.Context, jobID string, validated artifact.Manifest) error { | ||
| digested, digestErr := digestManifestFiles(validated) | ||
|
pullfrog[bot] marked this conversation as resolved.
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. WARNING: A digest failure during recovery aborts
Reply with |
||
| if digestErr != nil { | ||
| return fmt.Errorf("%w: %w", ErrJobOutputUnstable, digestErr) | ||
| } | ||
| tx, beginErr := store.database.BeginTx(ctx, nil) | ||
| if beginErr != nil { | ||
| return fmt.Errorf("begin job completion: %w", beginErr) | ||
|
|
@@ -249,7 +257,7 @@ func (store *Store) completeValidatedJob(ctx context.Context, jobID string, vali | |
| committed = true | ||
| return nil | ||
| } | ||
| if recordErr := recordManifestTx(ctx, tx, validated); recordErr != nil { | ||
| if recordErr := recordManifestTx(ctx, tx, digested); recordErr != nil { | ||
|
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. WARNING: No test asserts the digest persisted through The acceptance criterion ("recording stores SHA-256") is covered only for the Reply with |
||
| return recordErr | ||
| } | ||
| now := formatDatabaseTime(time.Now()) | ||
|
|
@@ -385,18 +393,35 @@ func (store *Store) RecoverInterruptedJobs(ctx context.Context, kind string) (Re | |
| continue | ||
| } | ||
| if err := store.completeValidatedJob(ctx, job.ID, manifest); err != nil { | ||
| if errors.Is(err, ErrJobTerminal) || errors.Is(err, ErrJobTransition) || errors.Is(err, ErrJobNotFound) { | ||
| result.Skipped = append(result.Skipped, job.ID) | ||
| continue | ||
| if containErr := store.containRecoveryError(ctx, &result, job.ID, err); containErr != nil { | ||
| return RecoveryResult{}, containErr | ||
| } | ||
| return RecoveryResult{}, err | ||
| continue | ||
| } | ||
| result.Recovered = append(result.Recovered, job.ID) | ||
| result.Artifacts = append(result.Artifacts, RecoveredArtifact{JobID: job.ID, Manifest: manifest}) | ||
| } | ||
| return result, nil | ||
| } | ||
|
|
||
| // containRecoveryError sorts one job's completion failure into the recovery | ||
| // result and returns nil when recovery can move on to the next job. Job-state | ||
| // failures skip the job; unstable outputs park it recoverable with a summary, | ||
| // the same containment the buildErr path above applies, so one unreadable | ||
| // output cannot wedge watch startup. Anything else is a durable-store failure | ||
| // the caller must abort on. | ||
| func (store *Store) containRecoveryError(ctx context.Context, result *RecoveryResult, jobID string, err error) error { | ||
| if errors.Is(err, ErrJobTerminal) || errors.Is(err, ErrJobTransition) || errors.Is(err, ErrJobNotFound) { | ||
| result.Skipped = append(result.Skipped, jobID) | ||
| return nil | ||
| } | ||
| if errors.Is(err, ErrJobOutputUnstable) { | ||
| result.Pending = append(result.Pending, jobID) | ||
| return store.setRecoverableSummary(ctx, jobID, "completed outputs changed or became unreadable before recording") | ||
| } | ||
| return err | ||
| } | ||
|
|
||
| // Job returns one durable job. | ||
| func (store *Store) Job(ctx context.Context, jobID string) (Job, error) { | ||
| if store == nil || store.database == nil { | ||
|
|
||
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
WARNING: A record-time digest failure fails the whole batch and, in
--eventsmode, reports fully published media as a failed jobRecordManifestsdigests every manifest beforeBeginTx, so one file that cannot be hashed (swap race in the window after the manifest was built, size drift, or EIO during the first full read) now leaves the entire run unrecorded:applyLibraryRecording(internal/cli/cli_download.go:271-279) setsLibraryRecorded=falsefor all lectures, anddownloadEventStream.finish(internal/cli/cli_download_events.go:47-49) converts that intoerrDownloadLibraryCommit— a terminaljob.failedevent and nonzero exit even though all media was published. Before this change, recording never read file bytes, so post-download filesystem state could not fail the commit (only DB errors could). If failing the record is the intended semantics (it prevents persisting a bogus reference digest), consider distinguishing "media published, library commit failed" from a download failure in events mode, or degrading per-manifest so one file does not suppress recording of the other lectures.Reply with
@kilocode-bot fix itto have Kilo Code address this issue.