diff --git a/AGENTS.md b/AGENTS.md index 2212f68..c087c7b 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -50,6 +50,8 @@ The working tree may already contain staged or unstaged changes. Preserve them a - Persist retry state before considering a rewrite or hardlink operation complete. - Preserve build-tagged behavior in `syscall_*`, `mmap/*`, and `cmd/acp-rewrite/file_*` files. - Do not add concurrency unless it has explicit ownership, completion, and error semantics. +- Keep the ACP content-signature xattr a disposable size-and-mtime cache. Transfers always hash content, managed cache keys are not copied as ordinary xattrs, and bounded cache writers drain before the pipeline returns. +- Report signature-cache read and write failures as aggregate warnings without adding them to `WaitErr`. - Add semantic regression tests for every behavior change. ## Verification diff --git a/README.md b/README.md index 8447092..2c4e49d 100644 --- a/README.md +++ b/README.md @@ -5,8 +5,15 @@ An Advanced Copy Tools, with following extra features: - Multi target path, read once write many - Read file with mmap, with small file prefetch hint - JSON format job report +- Optional SHA-256 xattr cache for hash-only streams - Can use as a golang library +## Content signature cache + +Library callers can opt in with `WithSignatureCache(true)` and bypass reads with `ForceRehash(true)`. ACP stores a fixed binary SHA-256, file size, and nanosecond mtime value in `user.acp.signature` on Linux and the canonical `acp.signature` user attribute on Darwin and FreeBSD. + +The xattr is a disposable optimization. Hash-only jobs may reuse a metadata-valid value. Jobs with targets always read and hash content, then refresh the source and every successful target before `WaitErr` or `RunStream` returns. Missing, stale, corrupt, read-only, full, or unsupported xattrs are summarized as warnings and never become copy errors. ACP's managed key is not copied as an ordinary source xattr. + # Install ``` # Install acp diff --git a/acp.go b/acp.go index d19b050..fffa920 100644 --- a/acp.go +++ b/acp.go @@ -18,6 +18,7 @@ type Copyer struct { getDevice func(in string) string getDiskUsageCache func(mountPoint string) *diskUsageCache linearTargetEnded uint32 + signatures *signatureCache } func New(ctx context.Context, opts ...Option) (*Copyer, error) { @@ -45,6 +46,9 @@ func New(ctx context.Context, opts ...Option) (*Copyer, error) { return newDiskUsageCache(mountPoint, defaultDiskUsageFreshInterval) }), } + if opt.withSignatureCache { + c.signatures = newSignatureCache(signatureWorkers(opt.fromDevice, opt.toDevice)) + } // Account for both pipeline and event dispatch before either goroutine starts. c.running.Add(2) @@ -93,6 +97,7 @@ func (c *Copyer) run(ctx context.Context) error { defer cancel() defer c.running.Done() defer close(c.eventCh) + defer c.finishSignatureCache() // Keep event dispatch alive until every pipeline stage stops publishing. go wrap(ctx, func() { c.eventLoop(ctx) }) diff --git a/cleanup.go b/cleanup.go index 602140b..8bded5b 100644 --- a/cleanup.go +++ b/cleanup.go @@ -29,6 +29,14 @@ func (c *Copyer) cleanupJob(ctx context.Context, cancel context.CancelFunc, copy } } + // Refresh only signatures backed by a complete source hash. + if c.signatures != nil && job.hashValid && !job.cacheHit { + c.signatures.enqueue(job.path, job.hash, job.stat) + for _, dst := range job.successTargets { + c.signatures.enqueue(dst, job.hash, nil) + } + } + // Publish only results whose data and metadata lifecycle has finished. job.setStatus(jobStatusFinished) if c.streamSink == nil { diff --git a/copy.go b/copy.go index c7a09e3..d99fa71 100644 --- a/copy.go +++ b/copy.go @@ -100,6 +100,11 @@ func (c *Copyer) write(ctx context.Context, job *writeJob, ch chan<- *baseJob, c job.fail("", fmt.Errorf("source size changed, indexed=%d current=%d", job.stat.size, job.size)) return } + if job.cacheHit { + atomic.AddInt64(&cntr.files, 1) + atomic.AddInt64(&cntr.bytes, job.size) + return + } // Skip jobs only when every requested target device is already exhausted. targetDevices := lo.Map(job.targets, func(target string, _ int) string { return c.getDevice(target) }) @@ -150,6 +155,10 @@ func (c *Copyer) write(ctx context.Context, job *writeJob, ch chan<- *baseJob, c continue } + // Invalidate an existing cache before O_TRUNC can replace its content. + if c.signatures != nil && c.createFlag&os.O_TRUNC != 0 { + c.signatures.invalidatePath(target) + } file, err := os.OpenFile(target, c.createFlag, job.stat.mode) if err = mappingError(err); err != nil { if checkErrorAbort(err) { @@ -160,6 +169,9 @@ func (c *Copyer) write(ctx context.Context, job *writeJob, ch chan<- *baseJob, c job.fail(target, fmt.Errorf("open dst file fail, %w", err)) continue } + if c.signatures != nil { + c.signatures.invalidate(file, target) + } if !job.copyer.toDevice.linear && job.size > 0 { if err := truncate(file, job.size); err != nil { _ = file.Close() @@ -270,6 +282,9 @@ func (c *Copyer) write(ctx context.Context, job *writeJob, ch chan<- *baseJob, c if readErr == nil && copied != job.size { readErr = fmt.Errorf("source size changed while copying, expected=%d copied=%d", job.size, copied) } + if readErr == nil && c.withHash { + job.validateHash() + } if readErr != nil && targetWriters == 0 { job.fail("", readErr) } diff --git a/event.go b/event.go index 3b99a1f..09f9883 100644 --- a/event.go +++ b/event.go @@ -32,6 +32,13 @@ type EventReportError struct { func (*EventReportError) iEvent() {} +// EventSignatureCacheSummary reports aggregate cache behavior for one Copyer. +type EventSignatureCacheSummary struct { + Summary SignatureCacheSummary +} + +func (*EventSignatureCacheSummary) iEvent() {} + type EventFinished struct{} func (*EventFinished) iEvent() {} diff --git a/job.go b/job.go index 27fdf21..0953a25 100644 --- a/job.go +++ b/job.go @@ -51,6 +51,8 @@ type baseJob struct { successTargets []string failedTargets map[string]error hash []byte + cacheHit bool + hashValid bool } func (j *baseJob) setStatus(s jobStatus) { @@ -73,6 +75,22 @@ func (j *baseJob) setHash(h []byte) { j.copyer.submit(&EventUpdateJob{j.report()}) } +func (j *baseJob) setCachedHash(h []byte) { + j.lock.Lock() + defer j.lock.Unlock() + + j.hash = h + j.cacheHit = true + j.hashValid = true + j.copyer.submit(&EventUpdateJob{j.report()}) +} + +func (j *baseJob) validateHash() { + j.lock.Lock() + j.hashValid = true + j.lock.Unlock() +} + func (j *baseJob) success(path string) { j.lock.Lock() defer j.lock.Unlock() @@ -116,6 +134,8 @@ func (j *baseJob) report() *Job { ModTime: j.stat.modTime, WriteTime: j.writeTime, SHA256: hex.EncodeToString(j.hash), + + SignatureCacheHit: j.cacheHit, } } @@ -173,4 +193,6 @@ type Job struct { ModTime time.Time `json:"mod_time"` WriteTime time.Time `json:"write_time"` SHA256 string `json:"sha256"` + + SignatureCacheHit bool `json:"signature_cache_hit,omitempty"` } diff --git a/opt.go b/opt.go index 5e783a7..d960c25 100644 --- a/opt.go +++ b/opt.go @@ -1,6 +1,7 @@ package acp import ( + "fmt" "os" "path/filepath" "strings" @@ -43,6 +44,9 @@ type option struct { createFlag int withHash bool + withSignatureCache bool + forceRehash bool + logger *logrus.Logger eventHanders []EventHandler } @@ -64,6 +68,12 @@ func (o *option) check() error { o.fromDevice.check() o.toDevice.check() + if o.withSignatureCache { + o.withHash = true + } + if o.forceRehash && !o.withSignatureCache { + return fmt.Errorf("force rehash requires signature cache") + } if o.logger == nil { o.logger = logrus.StandardLogger() } @@ -132,6 +142,23 @@ func WithHash(b bool) Option { } } +// WithSignatureCache enables content-signature reads and writes through xattrs. +// It also enables hashing so cache misses can be refreshed. +func WithSignatureCache(enabled bool) Option { + return func(o *option) *option { + o.withSignatureCache = enabled + return o + } +} + +// ForceRehash bypasses signature-cache reads while still refreshing the cache. +func ForceRehash(force bool) Option { + return func(o *option) *option { + o.forceRehash = force + return o + } +} + func WithLogger(logger *logrus.Logger) Option { return func(o *option) *option { o.logger = logger diff --git a/prepare.go b/prepare.go index ec4f99e..df66770 100644 --- a/prepare.go +++ b/prepare.go @@ -45,40 +45,55 @@ func (c *Copyer) prepare(ctx context.Context, indexed <-chan *baseJob) <-chan *w job.setStatus(jobStatusPreparing) - file, size, err := func(path string) (io.ReadCloser, int64, error) { - if c.fromDevice.linear { - file, err := os.Open(path) - if err != nil { - return nil, 0, fmt.Errorf("open src file fail, %w", err) - } - - fileInfo, err := file.Stat() - if err != nil { - _ = file.Close() - return nil, 0, fmt.Errorf("get src file stat fail, %w", err) - } - - return file, fileInfo.Size(), nil + // A targetless hash job may finish from a metadata-valid cache entry. + var file io.ReadCloser + var size int64 + if c.signatures != nil && !c.forceRehash && len(job.targets) == 0 { + if hash, ok := c.signatures.lookup(job.path, job.stat); ok { + job.setCachedHash(hash) + file = io.NopCloser(bytes.NewReader(nil)) + size = job.stat.size } + } - readerAt, err := mmap.Open(path) + // Cache misses and every transfer open the real source content. + if file == nil { + var err error + file, size, err = func(path string) (io.ReadCloser, int64, error) { + if c.fromDevice.linear { + file, err := os.Open(path) + if err != nil { + return nil, 0, fmt.Errorf("open src file fail, %w", err) + } + + fileInfo, err := file.Stat() + if err != nil { + _ = file.Close() + return nil, 0, fmt.Errorf("get src file stat fail, %w", err) + } + + return file, fileInfo.Size(), nil + } + + readerAt, err := mmap.Open(path) + if err != nil { + return nil, 0, fmt.Errorf("open src file by mmap fail, %w", err) + } + if readerAt.Len() == 0 { + if err := readerAt.Close(); err != nil { + return nil, 0, fmt.Errorf("close empty src file by mmap fail, %w", err) + } + return io.NopCloser(bytes.NewReader(nil)), 0, nil + } + + return mmap.NewReader(readerAt), int64(readerAt.Len()), nil + }(job.path) if err != nil { - return nil, 0, fmt.Errorf("open src file by mmap fail, %w", err) + c.reportError(job.path, "", err) + job.fail("", err) + job.setStatus(jobStatusFinished) + continue } - if readerAt.Len() == 0 { - if err := readerAt.Close(); err != nil { - return nil, 0, fmt.Errorf("close empty src file by mmap fail, %w", err) - } - return io.NopCloser(bytes.NewReader(nil)), 0, nil - } - - return mmap.NewReader(readerAt), int64(readerAt.Len()), nil - }(job.path) - if err != nil { - c.reportError(job.path, "", err) - job.fail("", err) - job.setStatus(jobStatusFinished) - continue } wj := newWriteJob(job, file, size, c.fromDevice.linear) diff --git a/signature.go b/signature.go new file mode 100644 index 0000000..5ab1255 --- /dev/null +++ b/signature.go @@ -0,0 +1,368 @@ +package acp + +import ( + "encoding/binary" + "errors" + "fmt" + "os" + "sync" + + "github.com/sirupsen/logrus" +) + +const ( + signatureMagic = "ACPS" + signatureVersion = uint8(1) + signatureSHA256 = uint8(1) + signatureEncodedSize = 56 + signatureQueueSize = 128 + signatureSampleLimit = 5 +) + +var errSignatureXattrUnsupported = errors.New("signature xattr is unsupported") + +// CachedSignature is a SHA-256 content signature bound to file metadata. +type CachedSignature struct { + Size int64 + MtimeNS int64 + SHA256 [32]byte +} + +// SignatureCacheSummary describes cache activity for one Copyer run. +type SignatureCacheSummary struct { + Hits int64 `json:"hits"` + Misses int64 `json:"misses"` + Stale int64 `json:"stale"` + Writes int64 `json:"writes"` + Failures int64 `json:"failures"` + FirstError string `json:"first_error,omitempty"` + Samples []string `json:"samples,omitempty"` +} + +func encodeCachedSignature(signature CachedSignature) []byte { + encoded := make([]byte, signatureEncodedSize) + copy(encoded[:4], signatureMagic) + encoded[4] = signatureVersion + encoded[5] = signatureSHA256 + binary.BigEndian.PutUint64(encoded[8:16], uint64(signature.Size)) + binary.BigEndian.PutUint64(encoded[16:24], uint64(signature.MtimeNS)) + copy(encoded[24:], signature.SHA256[:]) + return encoded +} + +// DecodeCachedSignature decodes ACP's stable, big-endian xattr representation. +func DecodeCachedSignature(encoded []byte) (CachedSignature, error) { + if len(encoded) != signatureEncodedSize { + return CachedSignature{}, fmt.Errorf("decode cached signature failed, size=%d", len(encoded)) + } + if string(encoded[:4]) != signatureMagic { + return CachedSignature{}, fmt.Errorf("decode cached signature failed, invalid magic") + } + if encoded[4] != signatureVersion { + return CachedSignature{}, fmt.Errorf("decode cached signature failed, version=%d", encoded[4]) + } + if encoded[5] != signatureSHA256 { + return CachedSignature{}, fmt.Errorf("decode cached signature failed, algorithm=%d", encoded[5]) + } + if encoded[6] != 0 || encoded[7] != 0 { + return CachedSignature{}, fmt.Errorf("decode cached signature failed, reserved bytes are not zero") + } + + var signature CachedSignature + signature.Size = int64(binary.BigEndian.Uint64(encoded[8:16])) + signature.MtimeNS = int64(binary.BigEndian.Uint64(encoded[16:24])) + copy(signature.SHA256[:], encoded[24:]) + return signature, nil +} + +type signatureReadStatus uint8 + +const ( + signatureReadMiss signatureReadStatus = iota + signatureReadHit + signatureReadStale +) + +// ReadCachedSignature returns a signature only when its size and mtime still +// match the regular file. Missing, stale, and unsupported xattrs are cache misses. +func ReadCachedSignature(path string) (CachedSignature, bool, error) { + signature, status, err := readCachedSignature(path) + if isSignatureXattrUnsupported(err) { + return CachedSignature{}, false, nil + } + return signature, status == signatureReadHit, err +} + +func readCachedSignature(path string) (CachedSignature, signatureReadStatus, error) { + file, err := os.Open(path) + if err != nil { + return CachedSignature{}, signatureReadMiss, fmt.Errorf("open signature source failed, %w", err) + } + defer file.Close() + + info, err := file.Stat() + if err != nil { + return CachedSignature{}, signatureReadMiss, fmt.Errorf("stat signature source failed, %w", err) + } + if !info.Mode().IsRegular() { + return CachedSignature{}, signatureReadMiss, fmt.Errorf("signature source is not a regular file") + } + + encoded, err := readSignatureXattr(file) + if err != nil { + if isSignatureXattrMissing(err) { + return CachedSignature{}, signatureReadMiss, nil + } + return CachedSignature{}, signatureReadMiss, fmt.Errorf("read signature xattr failed, %w", err) + } + signature, err := DecodeCachedSignature(encoded) + if err != nil { + return CachedSignature{}, signatureReadMiss, err + } + after, err := file.Stat() + if err != nil { + return CachedSignature{}, signatureReadMiss, fmt.Errorf("restat signature source failed, %w", err) + } + if !after.Mode().IsRegular() || after.Size() != info.Size() || !after.ModTime().Equal(info.ModTime()) { + return signature, signatureReadStale, nil + } + if signature.Size != info.Size() || signature.MtimeNS != info.ModTime().UnixNano() { + return signature, signatureReadStale, nil + } + return signature, signatureReadHit, nil +} + +type signatureWrite struct { + path string + signature CachedSignature +} + +type signatureCache struct { + queue chan signatureWrite + wg sync.WaitGroup + + lock sync.Mutex + summary SignatureCacheSummary +} + +func newSignatureCache(workers int) *signatureCache { + cache := &signatureCache{queue: make(chan signatureWrite, signatureQueueSize)} + cache.wg.Add(workers) + for idx := 0; idx < workers; idx++ { + go func() { + defer cache.wg.Done() + for write := range cache.queue { + cache.write(write) + } + }() + } + return cache +} + +func (c *signatureCache) lookup(path string, indexed *stat) ([]byte, bool) { + signature, status, err := readCachedSignature(path) + if err != nil { + c.recordFailure(path, err) + c.incrementMiss() + return nil, false + } + + switch status { + case signatureReadHit: + if signature.Size != indexed.size || signature.MtimeNS != indexed.modTime.UnixNano() { + c.incrementStale() + return nil, false + } + c.incrementHit() + return append([]byte(nil), signature.SHA256[:]...), true + case signatureReadStale: + c.incrementStale() + default: + c.incrementMiss() + } + return nil, false +} + +func (c *signatureCache) enqueue(path string, hash []byte, indexed *stat) { + 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) + c.queue <- signatureWrite{ + path: path, + signature: CachedSignature{ + Size: info.Size(), + MtimeNS: info.ModTime().UnixNano(), + SHA256: sum, + }, + } +} + +func (c *signatureCache) invalidate(file *os.File, path string) { + if err := removeSignatureXattr(file); err != nil && !isSignatureXattrMissing(err) { + c.recordFailure(path, fmt.Errorf("remove old signature xattr failed, %w", err)) + } +} + +func (c *signatureCache) invalidatePath(path string) { + file, err := os.Open(path) + if errors.Is(err, os.ErrNotExist) { + return + } + if err != nil { + c.recordFailure(path, fmt.Errorf("open old signature target failed, %w", err)) + return + } + defer file.Close() + c.invalidate(file, path) +} + +func (c *signatureCache) write(write signatureWrite) { + // Revalidate through the descriptor immediately before changing the xattr. + file, err := os.Open(write.path) + if err != nil { + c.recordFailure(write.path, fmt.Errorf("open signature target failed, %w", err)) + return + } + defer file.Close() + + info, err := file.Stat() + if err != nil { + c.recordFailure(write.path, fmt.Errorf("stat signature target failed, %w", err)) + return + } + if !info.Mode().IsRegular() { + c.recordFailure(write.path, fmt.Errorf("signature target is not a regular file")) + return + } + if info.Size() != write.signature.Size || info.ModTime().UnixNano() != write.signature.MtimeNS { + c.recordFailure(write.path, fmt.Errorf("signature target metadata changed")) + return + } + if err := writeSignatureXattr(file, encodeCachedSignature(write.signature)); err != nil { + c.recordFailure(write.path, fmt.Errorf("write signature xattr failed, %w", err)) + return + } + after, err := file.Stat() + if err != nil { + _ = removeSignatureXattr(file) + c.recordFailure(write.path, fmt.Errorf("restat signature target failed, %w", err)) + return + } + if after.Size() != write.signature.Size || after.ModTime().UnixNano() != write.signature.MtimeNS { + _ = removeSignatureXattr(file) + c.recordFailure(write.path, fmt.Errorf("signature target metadata changed during write")) + return + } + + c.lock.Lock() + c.summary.Writes++ + c.lock.Unlock() +} + +func (c *signatureCache) incrementHit() { + c.lock.Lock() + c.summary.Hits++ + c.lock.Unlock() +} + +func (c *signatureCache) incrementMiss() { + c.lock.Lock() + c.summary.Misses++ + c.lock.Unlock() +} + +func (c *signatureCache) incrementStale() { + c.lock.Lock() + c.summary.Stale++ + c.lock.Unlock() +} + +func (c *signatureCache) recordFailure(path string, err error) { + c.lock.Lock() + defer c.lock.Unlock() + + c.summary.Failures++ + if c.summary.FirstError == "" { + c.summary.FirstError = err.Error() + } + if len(c.summary.Samples) < signatureSampleLimit { + c.summary.Samples = append(c.summary.Samples, path) + } +} + +func (c *signatureCache) closeAndWait() SignatureCacheSummary { + close(c.queue) + c.wg.Wait() + + c.lock.Lock() + defer c.lock.Unlock() + c.summary.Samples = append([]string(nil), c.summary.Samples...) + return c.summary +} + +func (c *Copyer) finishSignatureCache() { + if c.signatures == nil { + return + } + + summary := c.signatures.closeAndWait() + c.submit(&EventSignatureCacheSummary{Summary: summary}) + level := logrus.InfoLevel + if summary.Failures > 0 { + level = logrus.WarnLevel + } + c.logf( + level, + "signature cache summary: hit=%d miss=%d stale=%d write=%d failure=%d first_error=%q samples=%q", + summary.Hits, + summary.Misses, + summary.Stale, + summary.Writes, + summary.Failures, + summary.FirstError, + summary.Samples, + ) +} + +func signatureWorkers(from, to *deviceOption) int { + if from.linear || to.linear { + return 1 + } + if from.threads > to.threads { + return from.threads + } + return to.threads +} + +func signatureCacheKey(key string) bool { + return key == "acp.signature" || key == "user.acp.signature" +} diff --git a/signature_test.go b/signature_test.go new file mode 100644 index 0000000..cc45f10 --- /dev/null +++ b/signature_test.go @@ -0,0 +1,383 @@ +package acp + +import ( + "bytes" + "context" + "crypto/sha256" + "encoding/hex" + "errors" + "fmt" + "os" + "path/filepath" + "testing" + "time" +) + +func TestCachedSignatureCodec(t *testing.T) { + want := CachedSignature{ + Size: 42, + MtimeNS: -123456789, + SHA256: sha256.Sum256([]byte("fixture")), + } + encoded := encodeCachedSignature(want) + got, err := DecodeCachedSignature(encoded) + if err != nil { + t.Fatal(err) + } + if got != want { + t.Fatalf("decoded signature = %#v, want %#v", got, want) + } + + tests := []struct { + name string + mutate func([]byte) []byte + }{ + {name: "size", mutate: func(value []byte) []byte { return value[:len(value)-1] }}, + {name: "magic", mutate: func(value []byte) []byte { value[0] ^= 0xff; return value }}, + {name: "version", mutate: func(value []byte) []byte { value[4]++; return value }}, + {name: "algorithm", mutate: func(value []byte) []byte { value[5]++; return value }}, + {name: "reserved", mutate: func(value []byte) []byte { value[6] = 1; return value }}, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + invalid := tt.mutate(append([]byte(nil), encoded...)) + if _, err := DecodeCachedSignature(invalid); err == nil { + t.Fatal("DecodeCachedSignature() error = nil") + } + }) + } +} + +func TestRunStreamSignatureCacheHitStaleAndForce(t *testing.T) { + content := []byte("cache fixture") + input := filepath.Join(t.TempDir(), "source.txt") + if err := os.WriteFile(input, content, 0o644); err != nil { + t.Fatal(err) + } + wantHash := sha256.Sum256(content) + + // The first hash populates the cache before RunStream returns. + first, firstSummary := runSignatureHash(t, input, false) + if first.SignatureCacheHit { + t.Fatal("first hash unexpectedly hit the signature cache") + } + if first.SHA256 != hex.EncodeToString(wantHash[:]) { + t.Fatalf("first SHA256 = %q, want %q", first.SHA256, hex.EncodeToString(wantHash[:])) + } + signature, valid, err := ReadCachedSignature(input) + if err != nil { + t.Fatal(err) + } + if !valid { + t.Skip("temporary filesystem does not support signature xattrs") + } + if signature.SHA256 != wantHash || firstSummary.Writes != 1 { + t.Fatalf("populated signature = %#v, summary = %#v", signature, firstSummary) + } + + // Unchanged metadata uses the cached SHA-256 without reading content. + second, secondSummary := runSignatureHash(t, input, false) + if !second.SignatureCacheHit || second.SHA256 != first.SHA256 { + t.Fatalf("cached job = %#v, want cache hit with SHA256 %q", second, first.SHA256) + } + if secondSummary.Hits != 1 || secondSummary.Writes != 0 { + t.Fatalf("cache-hit summary = %#v", secondSummary) + } + + // A metadata change makes the old signature stale and refreshes it. + info, err := os.Stat(input) + if err != nil { + t.Fatal(err) + } + changed := info.ModTime().Add(2 * time.Second) + if err := os.Chtimes(input, changed, changed); err != nil { + t.Fatal(err) + } + stale, staleSummary := runSignatureHash(t, input, false) + if stale.SignatureCacheHit || staleSummary.Stale != 1 || staleSummary.Writes != 1 { + t.Fatalf("stale job = %#v, summary = %#v", stale, staleSummary) + } + + // Force rehash bypasses the now-valid cache and still refreshes it. + forced, forcedSummary := runSignatureHash(t, input, true) + if forced.SignatureCacheHit || forcedSummary.Hits != 0 || forcedSummary.Writes != 1 { + t.Fatalf("forced job = %#v, summary = %#v", forced, forcedSummary) + } +} + +func TestRunStreamTransferAlwaysHashesAndRefreshesTargets(t *testing.T) { + content := []byte("transfer fixture") + root := t.TempDir() + input := filepath.Join(root, "source.txt") + target := filepath.Join(root, "target.txt") + if err := os.WriteFile(input, content, 0o644); err != nil { + t.Fatal(err) + } + if err := os.WriteFile(target, []byte("old"), 0o644); err != nil { + t.Fatal(err) + } + + // Seed a metadata-valid but incorrect cache entry to prove transfer ignores it. + info, err := os.Stat(input) + if err != nil { + t.Fatal(err) + } + wrong := CachedSignature{Size: info.Size(), MtimeNS: info.ModTime().UnixNano(), SHA256: sha256.Sum256([]byte("wrong"))} + file, err := os.Open(input) + if err != nil { + t.Fatal(err) + } + err = writeSignatureXattr(file, encodeCachedSignature(wrong)) + _ = file.Close() + if err != nil { + t.Skipf("temporary filesystem does not support signature xattrs: %v", err) + } + targetInfo, err := os.Stat(target) + if err != nil { + t.Fatal(err) + } + targetFile, err := os.Open(target) + if err != nil { + t.Fatal(err) + } + err = writeSignatureXattr(targetFile, encodeCachedSignature(CachedSignature{ + Size: targetInfo.Size(), MtimeNS: targetInfo.ModTime().UnixNano(), SHA256: sha256.Sum256([]byte("old")), + })) + _ = targetFile.Close() + if err != nil { + t.Skipf("temporary filesystem does not support target signature xattrs: %v", err) + } + + sink := new(collectingStreamSink) + var summary SignatureCacheSummary + handler := func(event Event) { + if update, ok := event.(*EventSignatureCacheSummary); ok { + summary = update.Summary + } + } + err = RunStream( + context.Background(), + &sliceStreamSource{requests: []*StreamRequest{{ID: 1, Source: input, Targets: []string{target}}}}, + sink, + Overwrite(true), + WithSignatureCache(true), + WithEventHandler(handler), + ) + if err != nil { + t.Fatal(err) + } + wantHash := sha256.Sum256(content) + job := sink.results[0].Job + if job.SignatureCacheHit || job.SHA256 != hex.EncodeToString(wantHash[:]) { + t.Fatalf("transfer job = %#v", job) + } + got, err := os.ReadFile(target) + if err != nil { + t.Fatal(err) + } + if !bytes.Equal(got, content) { + t.Fatalf("target content = %q, want %q", got, content) + } + for _, path := range []string{input, target} { + signature, valid, err := ReadCachedSignature(path) + if err != nil { + t.Fatal(err) + } + if !valid || signature.SHA256 != wantHash { + t.Fatalf("signature for %q = %#v, valid=%t", path, signature, valid) + } + } + if summary.Hits != 0 || summary.Writes != 2 { + t.Fatalf("transfer summary = %#v", summary) + } +} + +func TestRunStreamCorruptSignatureIsWarning(t *testing.T) { + input := filepath.Join(t.TempDir(), "source.txt") + if err := os.WriteFile(input, nil, 0o644); err != nil { + t.Fatal(err) + } + file, err := os.Open(input) + if err != nil { + t.Fatal(err) + } + err = writeSignatureXattr(file, []byte("corrupt")) + _ = file.Close() + if err != nil { + t.Skipf("temporary filesystem does not support signature xattrs: %v", err) + } + + job, summary := runSignatureHash(t, input, false) + if job.SignatureCacheHit || summary.Failures != 1 || summary.Misses != 1 || summary.Writes != 1 { + t.Fatalf("job = %#v, summary = %#v", job, summary) + } + if summary.FirstError == "" || len(summary.Samples) != 1 || summary.Samples[0] != input { + t.Fatalf("warning details = %#v", summary) + } +} + +func TestRunStreamSignatureCacheZeroLength(t *testing.T) { + input := filepath.Join(t.TempDir(), "empty") + if err := os.WriteFile(input, nil, 0o644); err != nil { + t.Fatal(err) + } + + first, _ := runSignatureHash(t, input, false) + if first.SHA256 != hex.EncodeToString(sha256.New().Sum(nil)) { + t.Fatalf("empty SHA256 = %q", first.SHA256) + } + _, valid, err := ReadCachedSignature(input) + if err != nil { + t.Fatal(err) + } + if !valid { + t.Skip("temporary filesystem does not support signature xattrs") + } + second, summary := runSignatureHash(t, input, false) + if !second.SignatureCacheHit || summary.Hits != 1 { + t.Fatalf("empty cache job = %#v, summary = %#v", second, summary) + } +} + +func TestSignatureCacheWarningSamplesAreBounded(t *testing.T) { + cache := newSignatureCache(1) + for idx := 0; idx < signatureSampleLimit+3; idx++ { + cache.recordFailure(fmt.Sprintf("path-%d", idx), errors.New("injected failure")) + } + summary := cache.closeAndWait() + if summary.Failures != signatureSampleLimit+3 { + t.Fatalf("failures = %d, want %d", summary.Failures, signatureSampleLimit+3) + } + if len(summary.Samples) != signatureSampleLimit { + t.Fatalf("samples = %d, want %d", len(summary.Samples), signatureSampleLimit) + } +} + +func TestRunStreamSignatureCacheDrainsConcurrentWrites(t *testing.T) { + root := t.TempDir() + source := new(sliceStreamSource) + want := make(map[string][32]byte) + for idx := 0; idx < 32; idx++ { + path := filepath.Join(root, fmt.Sprintf("%02d.bin", idx)) + content := []byte(fmt.Sprintf("concurrent signature %d", idx)) + if err := os.WriteFile(path, content, 0o644); err != nil { + t.Fatal(err) + } + source.requests = append(source.requests, &StreamRequest{ID: int64(idx + 1), Source: path}) + want[path] = sha256.Sum256(content) + } + + sink := new(collectingStreamSink) + if err := RunStream( + context.Background(), source, sink, + WithSignatureCache(true), + SetFromDevice(DeviceThreads(4)), + ); err != nil { + t.Fatal(err) + } + if len(sink.results) != len(want) { + t.Fatalf("received %d results, want %d", len(sink.results), len(want)) + } + for path, expected := range want { + signature, valid, err := ReadCachedSignature(path) + if err != nil { + t.Fatal(err) + } + if !valid { + t.Skip("temporary filesystem does not support signature xattrs") + } + if signature.SHA256 != expected { + t.Fatalf("signature for %q = %x, want %x", path, signature.SHA256, expected) + } + } +} + +type cancelingSignatureSink struct { + cancel context.CancelFunc + path string +} + +func (s *cancelingSignatureSink) Write(_ context.Context, result *StreamResult) error { + s.path = result.Job.FullPath + s.cancel() + return nil +} + +func (*cancelingSignatureSink) Flush(context.Context) error { + return nil +} + +func TestRunStreamSignatureCacheDrainsAfterCancellation(t *testing.T) { + path := filepath.Join(t.TempDir(), "cancellation.bin") + if err := os.WriteFile(path, []byte("cancellation fixture"), 0o644); err != nil { + t.Fatal(err) + } + + ctx, cancel := context.WithCancel(context.Background()) + sink := &cancelingSignatureSink{cancel: cancel} + err := RunStream(ctx, &blockingStreamSource{ + request: &StreamRequest{ID: 1, Source: path}, + }, sink, WithSignatureCache(true)) + if !errors.Is(err, context.Canceled) { + t.Fatalf("RunStream() error = %v, want context cancellation", err) + } + if sink.path == "" { + t.Fatal("cancellation sink received no completed result") + } + _, valid, err := ReadCachedSignature(sink.path) + if err != nil { + t.Fatal(err) + } + if !valid { + t.Skip("temporary filesystem does not support signature xattrs") + } +} + +func TestRunStreamSignatureCacheIsBestEffortForReadOnlyFile(t *testing.T) { + path := filepath.Join(t.TempDir(), "readonly.bin") + content := []byte("read-only signature fixture") + if err := os.WriteFile(path, content, 0o444); err != nil { + t.Fatal(err) + } + job, summary := runSignatureHash(t, path, true) + want := sha256.Sum256(content) + if job.SHA256 != hex.EncodeToString(want[:]) { + t.Fatalf("SHA256 = %q, want %q", job.SHA256, hex.EncodeToString(want[:])) + } + signature, valid, err := ReadCachedSignature(path) + if err != nil { + t.Fatal(err) + } + if valid && signature.SHA256 != want { + t.Fatalf("signature = %x, want %x", signature.SHA256, want) + } + if !valid && summary.Failures == 0 { + t.Fatalf("read-only cache miss had no warning: %#v", summary) + } +} + +func runSignatureHash(t *testing.T, input string, force bool) (*Job, SignatureCacheSummary) { + t.Helper() + + sink := new(collectingStreamSink) + var summary SignatureCacheSummary + handler := func(event Event) { + if update, ok := event.(*EventSignatureCacheSummary); ok { + summary = update.Summary + } + } + if err := RunStream( + context.Background(), + &sliceStreamSource{requests: []*StreamRequest{{ID: 1, Source: input}}}, + sink, + WithSignatureCache(true), + ForceRehash(force), + WithEventHandler(handler), + ); err != nil { + t.Fatal(err) + } + if len(sink.results) != 1 { + t.Fatalf("received %d results, want 1", len(sink.results)) + } + return sink.results[0].Job, summary +} diff --git a/signature_xattr_copy_test.go b/signature_xattr_copy_test.go new file mode 100644 index 0000000..6cfa8b9 --- /dev/null +++ b/signature_xattr_copy_test.go @@ -0,0 +1,43 @@ +//go:build darwin || linux +// +build darwin linux + +package acp + +import ( + "crypto/sha256" + "os" + "path/filepath" + "testing" +) + +func TestManagedSignatureIsNotOrdinaryCopiedXattr(t *testing.T) { + path := filepath.Join(t.TempDir(), "source.txt") + if err := os.WriteFile(path, []byte("fixture"), 0o644); err != nil { + t.Fatal(err) + } + info, err := os.Stat(path) + if err != nil { + t.Fatal(err) + } + file, err := os.Open(path) + if err != nil { + t.Fatal(err) + } + err = writeSignatureXattr(file, encodeCachedSignature(CachedSignature{ + Size: info.Size(), MtimeNS: info.ModTime().UnixNano(), SHA256: sha256.Sum256([]byte("fixture")), + })) + _ = file.Close() + if err != nil { + t.Skipf("temporary filesystem does not support signature xattrs: %v", err) + } + + xattrs, err := readXattrs(path) + if err != nil { + t.Fatal(err) + } + for _, xattr := range xattrs { + if signatureCacheKey(xattr.key) { + t.Fatalf("managed signature key %q was returned as an ordinary xattr", xattr.key) + } + } +} diff --git a/signature_xattr_darwin.go b/signature_xattr_darwin.go new file mode 100644 index 0000000..a0417e2 --- /dev/null +++ b/signature_xattr_darwin.go @@ -0,0 +1,6 @@ +//go:build darwin +// +build darwin + +package acp + +const signatureXattrName = "acp.signature" diff --git a/signature_xattr_freebsd.go b/signature_xattr_freebsd.go new file mode 100644 index 0000000..a516fd5 --- /dev/null +++ b/signature_xattr_freebsd.go @@ -0,0 +1,63 @@ +//go:build freebsd +// +build freebsd + +package acp + +import ( + "errors" + "os" + "unsafe" + + "golang.org/x/sys/unix" +) + +const signatureXattrName = "acp.signature" + +func readSignatureXattr(file *os.File) ([]byte, error) { + size, err := unix.ExtattrGetFd(int(file.Fd()), unix.EXTATTR_NAMESPACE_USER, signatureXattrName, 0, 0) + if err != nil { + return nil, err + } + value := make([]byte, size) + if size == 0 { + return value, nil + } + n, err := unix.ExtattrGetFd( + int(file.Fd()), + unix.EXTATTR_NAMESPACE_USER, + signatureXattrName, + uintptr(unsafe.Pointer(&value[0])), + len(value), + ) + if err != nil { + return nil, err + } + return value[:n], nil +} + +func writeSignatureXattr(file *os.File, value []byte) error { + var data uintptr + if len(value) > 0 { + data = uintptr(unsafe.Pointer(&value[0])) + } + _, err := unix.ExtattrSetFd( + int(file.Fd()), + unix.EXTATTR_NAMESPACE_USER, + signatureXattrName, + data, + len(value), + ) + return err +} + +func removeSignatureXattr(file *os.File) error { + return unix.ExtattrDeleteFd(int(file.Fd()), unix.EXTATTR_NAMESPACE_USER, signatureXattrName) +} + +func isSignatureXattrMissing(err error) bool { + return errors.Is(err, unix.ENOATTR) +} + +func isSignatureXattrUnsupported(err error) bool { + return errors.Is(err, unix.ENOTSUP) || errors.Is(err, unix.EOPNOTSUPP) +} diff --git a/signature_xattr_linux.go b/signature_xattr_linux.go new file mode 100644 index 0000000..0b0f3d7 --- /dev/null +++ b/signature_xattr_linux.go @@ -0,0 +1,6 @@ +//go:build linux +// +build linux + +package acp + +const signatureXattrName = "user.acp.signature" diff --git a/signature_xattr_other.go b/signature_xattr_other.go new file mode 100644 index 0000000..a564323 --- /dev/null +++ b/signature_xattr_other.go @@ -0,0 +1,31 @@ +//go:build !darwin && !freebsd && !linux +// +build !darwin,!freebsd,!linux + +package acp + +import ( + "errors" + "os" +) + +const signatureXattrName = "acp.signature" + +func readSignatureXattr(*os.File) ([]byte, error) { + return nil, errSignatureXattrUnsupported +} + +func writeSignatureXattr(*os.File, []byte) error { + return errSignatureXattrUnsupported +} + +func removeSignatureXattr(*os.File) error { + return errSignatureXattrUnsupported +} + +func isSignatureXattrMissing(err error) bool { + return false +} + +func isSignatureXattrUnsupported(err error) bool { + return errors.Is(err, errSignatureXattrUnsupported) +} diff --git a/signature_xattr_unix.go b/signature_xattr_unix.go new file mode 100644 index 0000000..9127b9e --- /dev/null +++ b/signature_xattr_unix.go @@ -0,0 +1,43 @@ +//go:build darwin || linux +// +build darwin linux + +package acp + +import ( + "errors" + "os" + + "golang.org/x/sys/unix" +) + +func readSignatureXattr(file *os.File) ([]byte, error) { + size, err := unix.Fgetxattr(int(file.Fd()), signatureXattrName, nil) + if err != nil { + return nil, err + } + value := make([]byte, size) + if size == 0 { + return value, nil + } + n, err := unix.Fgetxattr(int(file.Fd()), signatureXattrName, value) + if err != nil { + return nil, err + } + return value[:n], nil +} + +func writeSignatureXattr(file *os.File, value []byte) error { + return unix.Fsetxattr(int(file.Fd()), signatureXattrName, value, 0) +} + +func removeSignatureXattr(file *os.File) error { + return unix.Fremovexattr(int(file.Fd()), signatureXattrName) +} + +func isSignatureXattrMissing(err error) bool { + return isNoAttrErr(err) +} + +func isSignatureXattrUnsupported(err error) bool { + return errors.Is(err, unix.ENOTSUP) || errors.Is(err, unix.EOPNOTSUPP) +} diff --git a/syscall_darwin.go b/syscall_darwin.go index 20fd40a..3af784c 100644 --- a/syscall_darwin.go +++ b/syscall_darwin.go @@ -23,5 +23,5 @@ func isNoAttrErr(err error) bool { } func checkXattrKey(key string) bool { - return !strings.HasPrefix(key, "system.") + return !strings.HasPrefix(key, "system.") && !signatureCacheKey(key) } diff --git a/syscall_linux.go b/syscall_linux.go index 358ba0e..c2b0767 100644 --- a/syscall_linux.go +++ b/syscall_linux.go @@ -23,5 +23,5 @@ func isNoAttrErr(err error) bool { } func checkXattrKey(key string) bool { - return true + return !signatureCacheKey(key) }