diff --git a/AGENTS.md b/AGENTS.md index 0298005..509db75 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -88,7 +88,12 @@ entry point is `NewStream` with `Submit`/`Close`/`Wait`; `Run` and `BatchSource` reads the content, and the computed hash is published through it, synchronously, after the hash is complete and before the descriptor closes. A target publishes its entry through the descriptor the writer wrote it with, while that descriptor is still open; no cache operation - reopens a path, and no cache write outlives its item. + reopens a path and no cache write outlives its item, except for dropping a stale target entry by + path before that target is truncated, which happens before the target has a descriptor at all. +- A cache entry is published only for the file version whose bytes produced the hash: its size and + mtime are what a later lookup compares, so a source that cannot be shown, through the descriptor + the item still owns, to hold the facts that version had publishes nothing instead of binding the + computed hash to metadata it never saw. - Repeated value options are last-wins (`WithEventHandler` included), and the shell rejects a repeated relative path instead of copying one target twice. diff --git a/README.md b/README.md index 9ffe474..901dc49 100644 --- a/README.md +++ b/README.md @@ -268,13 +268,17 @@ entry is published through the descriptor that wrote it, while that descriptor i one path-based step runs before a target has a descriptor at all: an overwrite drops the target's stale entry before the file is truncated, so a crash cannot leave the old entry describing the new bytes. -Writers publish the hashed version's size and mtime unchanged, so an entry whose file changed -after hashing simply fails the reader's metadata comparison and is treated as stale. Missing, -stale, corrupt, read-only, full, or unsupported xattrs are summarized as warnings and never -become copy errors. A file system without the managed attribute namespace stores no cache at -all, which is a no-op: reading it is a miss and writing or removing it records no failure. -ACP's managed key is not copied as an ordinary source xattr. The run reports its aggregate -summary as `EventSignatureCacheSummary`. +Writers publish an entry only for the version whose bytes produced the hash, and they state that +version's size and mtime. The evidence is the descriptor the item still owns: a source keeps its +own metadata, so it must still show the facts the item observed, and a source that cannot be shown +to be that version publishes nothing instead of binding its hash to metadata the run never saw. A +target receives the item's metadata after the entry is published, so its length is what the +descriptor has to show. An entry whose file changes after it was written fails the reader's +metadata comparison and is treated as stale. Missing, stale, corrupt, read-only, full, or +unsupported xattrs are summarized as warnings and never become copy errors. A file system without +the managed attribute namespace stores no cache at all, which is a no-op: reading it is a miss and +writing or removing it records no failure. ACP's managed key is not copied as an ordinary source +xattr. The run reports its aggregate summary as `EventSignatureCacheSummary`. ## Platform behaviour diff --git a/copy.go b/copy.go index 9dbf214..1f53b5f 100644 --- a/copy.go +++ b/copy.go @@ -132,8 +132,10 @@ func (c *StreamCopyer) write(ctx context.Context, job *writeJob, ch chan<- *base wg.Wait() // The item publishes its source's computed signature through the descriptor that read the - // content, before that descriptor closes and before the result is published. - c.refreshCacheEntry(job.source, job.path, job.baseJob) + // content, before that descriptor closes and before the result is published. The source is + // not a file this run wrote, so the descriptor has to still show the facts the item + // observed for the entry to describe the bytes that were hashed. + c.refreshCacheEntry(job.source, job.path, job.baseJob, false) job.finishSource() c.publish(ch, job.baseJob) @@ -321,11 +323,13 @@ func (c *StreamCopyer) write(ctx context.Context, job *writeJob, ch chan<- *base // This target's content is complete, but the item's hash is only complete once the // whole source was read. The writer waits for it and publishes the target's cache - // entry while it still owns the descriptor it wrote through. + // entry while it still owns the descriptor it wrote through. The target holds exactly + // the bytes that produced the hash and is stamped with the item's metadata after this + // publication, so its length is what the descriptor has to show. if cacheGate != nil { <-cacheGate } - c.refreshCacheEntry(file, target, job.baseJob) + c.refreshCacheEntry(file, target, job.baseJob, true) if err := file.Close(); err != nil { file = nil diff --git a/signature.go b/signature.go index 76ed287..b52bfac 100644 --- a/signature.go +++ b/signature.go @@ -160,7 +160,9 @@ func readCachedSignatureFile(file *os.File) (CachedSignature, signatureReadStatu // signatureCache is the aggregate accounting of one run's cache activity. It owns no writer and // no queue: an item reads its stored hash and publishes its computed one through the descriptor it -// already owns, so a cache operation never reopens a path and never outlives its item. +// already owns, so a cache operation never reopens a path and never outlives its item. Dropping a +// target's stale entry before that target is truncated is the one path-based cache step, and it +// happens before the target has a descriptor at all. type signatureCache struct { lock sync.Mutex summary SignatureCacheSummary @@ -240,7 +242,22 @@ func (c *StreamCopyer) invalidateSignaturePath(path string) { // descriptor that owns one of that item's files. It publishes nothing when the item has no hash it // computed itself or the policy does not refresh the cache, and a nil descriptor is an item that // never opened the file it names rather than a failure. -func (c *StreamCopyer) refreshCacheEntry(file *os.File, path string, job *baseJob) { +// +// An entry states the size and modification time a later lookup compares, so it may only be +// published for the exact file version whose bytes produced the hash: an entry that binds that hash +// to other facts makes a later run report a hash for content it never read. The descriptor the item +// still owns is the evidence, and which evidence it can give depends on which of the item's files +// it is: +// +// - The source descriptor must still show the size the item read and the modification time it +// indexed. The run never wrote that file, so a source whose metadata moved after indexing cannot +// be shown to be the version the run hashed, and it publishes nothing instead of a guess. +// - A target descriptor is a file this run wrote with exactly the bytes that were hashed, and the +// run stamps the item's metadata onto it after this publication, so its length is the evidence +// available here. +// +// Every check runs through that descriptor, so a cache operation never reopens a path. +func (c *StreamCopyer) refreshCacheEntry(file *os.File, path string, job *baseJob, target bool) { if file == nil || c.signatures == nil || !c.hashPolicy.refreshesCache() { return } @@ -255,6 +272,15 @@ func (c *StreamCopyer) refreshCacheEntry(file *os.File, path string, job *baseJo return } + info, err := file.Stat() + if err != nil { + c.signatures.recordFailure(path, fmt.Errorf("stat signature target failed, %w", err)) + return + } + if info.Size() != signature.Size || (!target && info.ModTime().UnixNano() != signature.MtimeNS) { + return + } + c.signatures.refresh(file, path, signature) } diff --git a/signature_test.go b/signature_test.go index 96e967a..3b38133 100644 --- a/signature_test.go +++ b/signature_test.go @@ -264,6 +264,102 @@ func TestRunRefreshWritesOnlyWhenStoredHashDiffers(t *testing.T) { } } +// TestRunPublishesNoEntryWhenTheSourceChangedAfterIndexing pins the facts a refreshed entry may +// state. An entry's size and modification time are what a later lookup compares, so publishing one +// for a version the run cannot show it hashed makes that lookup answer for bytes it never read. The +// previous revision bound the bytes it read to the modification time it indexed, so restoring the +// first version with its original metadata produced a cache hit reporting the second version's +// hash. +func TestRunPublishesNoEntryWhenTheSourceChangedAfterIndexing(t *testing.T) { + requireSignatureXattrSupport(t) + + v1 := []byte("AAAAAAAAAAAAAAAA") + v2 := []byte("BBBBBBBBBBBBBBBB") + root := t.TempDir() + input := writeSourceFile(t, root, "source.txt", v1) + + indexed := time.Unix(1000, 0) + replaced := time.Unix(2000, 0) + if err := os.Chtimes(input, indexed, indexed); err != nil { + t.Fatal(err) + } + + // Indexing stats the path when the item is submitted, and the item opens its one descriptor + // during preparation, so this replacement lands after indexing and before the read. + previous := openSourceContent + openSourceContent = func(name string, mode ReadMode) (itemSource, error) { + source, err := previous(name, mode) + if err == nil && name == input { + if err := os.WriteFile(input, v2, 0o644); err != nil { + t.Errorf("replace the source: %v", err) + } + if err := os.Chtimes(input, replaced, replaced); err != nil { + t.Errorf("restamp the source: %v", err) + } + } + return source, err + } + t.Cleanup(func() { openSourceContent = previous }) + + item := newFixtureItem(input) + if err := runFixture(context.Background(), newStreamFixture(item), []Item{item}, WithHashPolicy(HashReadRefresh)); err != nil { + t.Fatalf("run error = %v", err) + } + openSourceContent = previous + + result, err := item.terminal(t) + if err != nil { + t.Fatalf("item failed: %v", err) + } + // The item still reports the bytes it read: a source that changed while a run is in progress + // is out of scope, and the run never fails the item for it. + want := sha256.Sum256(v2) + if !bytes.Equal(result.SHA256, want[:]) { + t.Fatalf("result SHA256 = %x, want the bytes it read %x", result.SHA256, want) + } + if stored, valid := readStoredSignature(t, input); valid { + t.Fatalf("published %#v for a source whose metadata moved after indexing", stored) + } + + // A restore puts the first version back with its original metadata. That is exactly the version + // the poisoned entry used to describe, so it has to read as a miss instead. + if err := os.WriteFile(input, v1, 0o644); err != nil { + t.Fatal(err) + } + if err := os.Chtimes(input, indexed, indexed); err != nil { + t.Fatal(err) + } + restored := sha256.Sum256(v1) + reuse := newFixtureItem(input) + if err := runFixture(context.Background(), newStreamFixture(reuse), []Item{reuse}, WithHashPolicy(HashCachedOrRead)); err != nil { + t.Fatal(err) + } + reused, err := reuse.terminal(t) + if err != nil { + t.Fatalf("restored item failed: %v", err) + } + if reused.SignatureCacheHit || !bytes.Equal(reused.SHA256, restored[:]) { + t.Fatalf("restored item = hit=%t sha256=%x, want a read of the restored content %x", reused.SignatureCacheHit, reused.SHA256, restored) + } + + // A consistent item still publishes, and the entry states the facts of the file it read. + item = newFixtureItem(input) + if err := runFixture(context.Background(), newStreamFixture(item), []Item{item}, WithHashPolicy(HashReadRefresh)); err != nil { + t.Fatal(err) + } + if _, err := item.terminal(t); err != nil { + t.Fatalf("consistent item failed: %v", err) + } + info, err := os.Stat(input) + if err != nil { + t.Fatal(err) + } + stored, valid := readStoredSignature(t, input) + if !valid || stored.Size != info.Size() || stored.MtimeNS != info.ModTime().UnixNano() || stored.SHA256 != restored { + t.Fatalf("consistent run stored %#v, valid=%t, want the facts of the file it read", stored, valid) + } +} + func TestRunTransferAlwaysReadsAndRefreshesTargets(t *testing.T) { // Prepare an existing target so the transfer exercises overwrite invalidation. content := []byte("transfer fixture")