acp/signature.go

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) {
	// Reject any representation outside the one stable codec version.
	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")
	}

	// Decode fixed-width facts only after the complete header is valid.
	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) {
	// Bind metadata and xattr reads to one regular-file descriptor.
	file, err := os.Open(path)
	if err != nil {
		return CachedSignature{}, signatureReadMiss, fmt.Errorf("open signature source failed, %w", err)
	}
	defer file.Close()

	// Capture the regular-file facts that bind the cached signature.
	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")
	}

	// Decode the managed xattr while preserving miss and failure semantics.
	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
	}

	// Revalidate metadata after the xattr read before declaring a cache hit.
	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 {
	// Start the bounded writer pool before any copy stage can enqueue work.
	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) {
	// Treat every unusable cache read as a non-fatal miss with diagnostics.
	signature, status, err := readCachedSignature(path)
	if err != nil {
		c.recordFailure(path, err)
		c.incrementMiss()
		return nil, false
	}

	// Distinguish metadata staleness from an absent entry for aggregate reporting.
	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) {
	// 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 *Copyer) invalidateSignature(file *os.File, path string) {
	if err := removeSignatureXattr(file); err != nil && !isSignatureXattrMissing(err) {
		if c.signatures != nil {
			c.signatures.recordFailure(path, fmt.Errorf("remove old signature xattr failed, %w", err))
		}
	}
}

func (c *Copyer) invalidateSignaturePath(path string) {
	file, err := os.Open(path)
	if errors.Is(err, os.ErrNotExist) {
		return
	}
	if err != nil {
		if c.signatures != nil {
			c.signatures.recordFailure(path, fmt.Errorf("open old signature target failed, %w", err))
		}
		return
	}
	defer file.Close()
	c.invalidateSignature(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()

	// Verify the queued metadata snapshot before changing the xattr.
	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
	}

	// Publish the xattr, then remove it if metadata changed during the write.
	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
	}

	// Count only signatures that survived the complete publication check.
	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 {
	// Stop accepting writes and drain every queued filesystem operation.
	close(c.queue)
	c.wg.Wait()

	// Return an isolated diagnostic snapshot after all workers have stopped.
	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
	}

	// Drain all best-effort writes before reporting the final aggregate summary.
	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"
}