mirrors/acp
1
0
Fork 0
mirror of https://github.com/samuelncui/acp.git synced 2026-09-03 14:47:17 +00:00
Code Issues Packages Projects Releases Wiki Activity

fix: harden signature cache publication

Browse Source
This commit is contained in:
Samuel Cui
2026-08-31 22:36:12 +08:00
parent cc5248ea33
commit d097ca107d
7 changed files with 132 additions and 67 deletions
Download Patch File Download Diff File Expand all files Collapse all files
cleanup.go
+8 -3
View File
@@ -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)
}
}
}
opt.go
+6 -2
View File
@@ -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
}
opt_device.go
+7 -1
View File
@@ -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
opt_test.go
+39
View File
@@ -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)
}
})
}
}
prepare.go
+17 -17
View File
@@ -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
}
}
signature.go
+19 -44
View File
@@ -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) {
signature_test.go
+36
View File
@@ -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")