From d097ca107d5f2f96f758545f3a9f310ca5796e46 Mon Sep 17 00:00:00 2001 From: Samuel Cui Date: Mon, 31 Aug 2026 22:36:12 +0800 Subject: [PATCH] fix: harden signature cache publication --- cleanup.go | 11 ++++++--- opt.go | 8 ++++-- opt_device.go | 8 +++++- opt_test.go | 39 +++++++++++++++++++++++++++++ prepare.go | 34 ++++++++++++------------- signature.go | 63 ++++++++++++++--------------------------------- signature_test.go | 36 +++++++++++++++++++++++++++ 7 files changed, 132 insertions(+), 67 deletions(-) diff --git a/cleanup.go b/cleanup.go index dddb894..b5584ae 100644 --- a/cleanup.go +++ b/cleanup.go @@ -32,9 +32,14 @@ func (c *Copyer) cleanupJob(ctx context.Context, cancel context.CancelFunc, copy // Refresh only signatures backed by a complete source hash. shouldRefreshSignature := c.signatures != nil && job.hashValid && !job.cacheHit if shouldRefreshSignature { - c.signatures.enqueue(job.path, job.hash, job.stat) - for _, dst := range job.successTargets { - c.signatures.enqueue(dst, job.hash, nil) + signature, err := newCachedSignature(job.hash, job.stat) + if err != nil { + c.signatures.recordFailure(job.path, err) + } else { + c.signatures.enqueue(job.path, signature) + for _, dst := range job.successTargets { + c.signatures.enqueue(dst, signature) + } } } diff --git a/opt.go b/opt.go index d960c25..5077b86 100644 --- a/opt.go +++ b/opt.go @@ -66,8 +66,12 @@ func (o *option) check() error { } } - o.fromDevice.check() - o.toDevice.check() + if err := o.fromDevice.check(); err != nil { + return fmt.Errorf("check source device failed, %w", err) + } + if err := o.toDevice.check(); err != nil { + return fmt.Errorf("check target device failed, %w", err) + } if o.withSignatureCache { o.withHash = true } diff --git a/opt_device.go b/opt_device.go index 8903170..878180b 100644 --- a/opt_device.go +++ b/opt_device.go @@ -1,17 +1,23 @@ package acp +import "fmt" + type deviceOption struct { linear bool threads int } -func (do *deviceOption) check() { +func (do *deviceOption) check() error { + if do.threads < 0 { + return fmt.Errorf("device threads cannot be negative, threads=%d", do.threads) + } if do.threads == 0 { do.threads = 8 } if do.linear { do.threads = 1 } + return nil } type DeviceOption func(*deviceOption) *deviceOption diff --git a/opt_test.go b/opt_test.go index 193a4f9..c878cfd 100644 --- a/opt_test.go +++ b/opt_test.go @@ -1,8 +1,10 @@ package acp import ( + "context" "path/filepath" "sort" + "strings" "testing" ) @@ -98,3 +100,40 @@ func TestLinearDeviceOnlySerializesItsOwnStage(t *testing.T) { }) } } + +func TestNewRejectsNegativeDeviceThreads(t *testing.T) { + tests := []struct { + name string + option Option + want string + }{ + { + name: "source", + option: SetFromDevice(DeviceThreads(-1)), + want: "check source device failed", + }, + { + name: "target", + option: SetToDevice(DeviceThreads(-1)), + want: "check target device failed", + }, + { + name: "linear source", + option: SetFromDevice(DeviceThreads(-1), LinearDevice(true)), + want: "check source device failed", + }, + } + + for _, test := range tests { + t.Run(test.name, func(t *testing.T) { + copyer, err := New(context.Background(), test.option) + if err == nil { + copyer.Wait() + t.Fatal("New() error = nil") + } + if !strings.Contains(err.Error(), test.want) || !strings.Contains(err.Error(), "threads=-1") { + t.Fatalf("New() error = %q, want %q and thread count", err, test.want) + } + }) + } +} diff --git a/prepare.go b/prepare.go index 3cd0ef4..fb7a037 100644 --- a/prepare.go +++ b/prepare.go @@ -26,16 +26,16 @@ func (c *Copyer) prepare(ctx context.Context, indexed <-chan *baseJob) <-chan *w defer close(ch) wg.Wait() }) - }() + }() // Prepare source readers with the configured source-device concurrency. for idx := 0; idx < c.fromDevice.threads; idx++ { wg.Add(1) - go wrap(ctx, func() { - defer wg.Done() + go wrap(ctx, func() { + defer wg.Done() - // Consume indexed jobs until cancellation or source exhaustion. - for { + // Consume indexed jobs until cancellation or source exhaustion. + for { select { case <-ctx.Done(): return @@ -44,12 +44,12 @@ func (c *Copyer) prepare(ctx context.Context, indexed <-chan *baseJob) <-chan *w return } if c.linearTargetStopped() { - continue - } + continue + } - // Enter preparation and let eligible targetless jobs reuse a valid cache entry. - job.setStatus(jobStatusPreparing) - var file io.ReadCloser + // Enter preparation and let eligible targetless jobs reuse a valid cache entry. + job.setStatus(jobStatusPreparing) + var file io.ReadCloser var size int64 cacheEligible := c.signatures != nil && !c.forceRehash && len(job.targets) == 0 if cacheEligible { @@ -96,19 +96,19 @@ func (c *Copyer) prepare(ctx context.Context, indexed <-chan *baseJob) <-chan *w c.reportError(job.path, "", err) job.fail("", err) job.setStatus(jobStatusFinished) - continue - } + continue } + } - // Publish the prepared reader and preserve linear-source consumption order. - wj := newWriteJob(job, file, size, c.fromDevice.linear) + // Publish the prepared reader and preserve linear-source consumption order. + wj := newWriteJob(job, file, size, c.fromDevice.linear) select { case ch <- wj: case <-ctx.Done(): wj.finishSource() - return - } - if !wj.waitConsumed(ctx) { + return + } + if !wj.waitConsumed(ctx) { return } } diff --git a/signature.go b/signature.go index c78aac0..2864b6b 100644 --- a/signature.go +++ b/signature.go @@ -167,6 +167,22 @@ func newSignatureCache(workers int) *signatureCache { return cache } +func newCachedSignature(hash []byte, indexed *stat) (CachedSignature, error) { + if len(hash) != len(CachedSignature{}.SHA256) { + return CachedSignature{}, fmt.Errorf("invalid SHA-256 size=%d", len(hash)) + } + if indexed == nil { + return CachedSignature{}, fmt.Errorf("signature metadata is missing") + } + + signature := CachedSignature{ + Size: indexed.size, + MtimeNS: indexed.modTime.UnixNano(), + } + copy(signature.SHA256[:], hash) + return signature, nil +} + func (c *signatureCache) lookup(path string, indexed *stat) ([]byte, bool) { // Treat every unusable cache read as a non-fatal miss with diagnostics. signature, status, err := readCachedSignature(path) @@ -193,50 +209,9 @@ func (c *signatureCache) lookup(path string, indexed *stat) ([]byte, bool) { return nil, false } -func (c *signatureCache) enqueue(path string, hash []byte, indexed *stat) { - // Accept only complete SHA-256 facts from the finished copy pipeline. - if len(hash) != len(CachedSignature{}.SHA256) { - c.recordFailure(path, fmt.Errorf("invalid SHA-256 size=%d", len(hash))) - return - } - - // Capture the metadata state that the asynchronous worker must preserve. - file, err := os.Open(path) - if err != nil { - c.recordFailure(path, fmt.Errorf("open signature target failed, %w", err)) - return - } - info, statErr := file.Stat() - closeErr := file.Close() - if statErr != nil { - c.recordFailure(path, fmt.Errorf("stat signature target failed, %w", statErr)) - return - } - if closeErr != nil { - c.recordFailure(path, fmt.Errorf("close signature target failed, %w", closeErr)) - return - } - if !info.Mode().IsRegular() { - c.recordFailure(path, fmt.Errorf("signature target is not a regular file")) - return - } - if indexed != nil && (info.Size() != indexed.size || info.ModTime().UnixNano() != indexed.modTime.UnixNano()) { - c.recordFailure(path, fmt.Errorf("signature source metadata changed")) - return - } - - var sum [32]byte - copy(sum[:], hash) - - // Queue an immutable signature snapshot for descriptor-level revalidation. - c.queue <- signatureWrite{ - path: path, - signature: CachedSignature{ - Size: info.Size(), - MtimeNS: info.ModTime().UnixNano(), - SHA256: sum, - }, - } +func (c *signatureCache) enqueue(path string, signature CachedSignature) { + // Queue the ACP result unchanged for descriptor-level revalidation. + c.queue <- signatureWrite{path: path, signature: signature} } func (c *Copyer) invalidateSignature(file *os.File, path string) { diff --git a/signature_test.go b/signature_test.go index 2cdec0d..d2790cf 100644 --- a/signature_test.go +++ b/signature_test.go @@ -50,6 +50,42 @@ func TestCachedSignatureCodec(t *testing.T) { } } +func TestSignatureCacheWritePreservesACPSnapshot(t *testing.T) { + // Build a signature from ACP's completed result rather than live file metadata. + content := []byte("snapshot fixture") + indexed := &stat{ + size: int64(len(content)), + modTime: time.Unix(100, 123), + } + hash := sha256.Sum256(content) + want, err := newCachedSignature(hash[:], indexed) + if err != nil { + t.Fatal(err) + } + + // Enqueueing must preserve that immutable snapshot even when the path differs. + path := filepath.Join(t.TempDir(), "changed.bin") + if err := os.WriteFile(path, bytes.Repeat([]byte("x"), len(content)), 0o644); err != nil { + t.Fatal(err) + } + changed := time.Unix(200, 456) + if err := os.Chtimes(path, changed, changed); err != nil { + t.Fatal(err) + } + cache := &signatureCache{queue: make(chan signatureWrite, 1)} + cache.enqueue(path, want) + write := <-cache.queue + if write.signature != want { + t.Fatalf("queued signature = %#v, want %#v", write.signature, want) + } + + // The writer must reject a path that no longer matches ACP's result. + cache.write(write) + if cache.summary.Writes != 0 || cache.summary.Failures != 1 { + t.Fatalf("signature summary = %#v", cache.summary) + } +} + func TestRunStreamSignatureCacheHitStaleAndForce(t *testing.T) { // Create one stable source whose cache lifecycle can be observed across runs. content := []byte("cache fixture")