Files
acp/fs.go
T
Samuel Cui ac02c71d3f feat!: stream caller-owned items and own the result queue
Reshape the public API around one push engine and restore the af05f05c
surface as a compatibility shell over it.

- NewStream takes caller-owned items (Source/Targets), reports every accepted
  item through one results callback, and Wait returns the run's terminal error.
  Run, BatchSource and the item callbacks are gone.
- The engine owns the result path: WithResultBuffer is the depth of the result
  queue, WithResultBatch is how many results one delivery carries, and
  WithResultFlushInterval is how long a successful result may wait. A result
  that carries any error is delivered immediately as its own batch.
- Restore the af05f05c API byte-compatibly in the same package: New, Copyer,
  Wait/WaitErr, AccurateJob, WildcardJob, the report events and the JSON report
  round-trip through encoding/json alone.
- One descriptor serves a whole item; the stored hash is read through the
  descriptor that reads the content and the computed hash is published through
  it before it closes.
- Make every handoff escapable by a hard stop: the read buffer, a chunk send,
  an event send and the result buffer.
- Fix the recorded sweep findings: report indent and round-trip, cmd/acp exit
  status, acp-rewrite final paths, mmap EOF/close/slice, disk-usage
  reservations, ErrTargetIO identity, the fatal-panic safety net, option
  validation, and the FreeBSD/Windows build paths.
- Cover the contract with semantic tests, add Makefile check/race/vet/cross
  targets, and document the withdrawn v0.1.0/v0.2.0 migration in README.
2026-09-19 15:18:13 +08:00

92 lines
2.4 KiB
Go

package acp
import (
"errors"
"fmt"
"io/fs"
"os"
"path/filepath"
"strings"
mapset "github.com/deckarep/golang-set/v2"
"github.com/moby/sys/mountinfo"
)
// absPath resolves a path against the current working directory. It is a variable so a test
// can force the failure path that a real working directory almost never produces.
var absPath = filepath.Abs
// getMountpointCache returns a memoized resolver from a path to the longest mount point that
// contains it.
func getMountpointCache() (func(string) (string, error), error) {
mounts, err := mountinfo.GetMounts(nil)
if err != nil {
return nil, fmt.Errorf("get mounts fail, %w", err)
}
mountPoints := mapset.NewThreadUnsafeSet[string]()
for _, mount := range mounts {
if mount == nil {
continue
}
if mount.Mountpoint == "" {
continue
}
mountPoints.Add(filepath.Clean(mount.Mountpoint))
}
mps := mountPoints.ToSlice()
resolve := Cache(func(path string) mountPoint {
// A path that cannot be made absolute has no mount point to report: the caller
// receives the failure instead of a panic it cannot recover from.
abs, err := absPath(path)
if err != nil {
return mountPoint{err: fmt.Errorf("get abs from file path failed, path= '%s', %w", path, err)}
}
return mountPoint{point: findMountpoint(abs, mps)}
})
return func(path string) (string, error) {
matched := resolve(path)
return matched.point, matched.err
}, nil
}
// mountPoint is the memoized outcome of resolving one path to its mount point. Both the match
// and the failure are cached, because a path that cannot be resolved cannot become resolvable.
type mountPoint struct {
point string
err error
}
// openSource opens a source for buffered reading without updating its access time.
// O_NOATIME needs ownership or privilege, so a refused open falls back to an ordinary one.
func openSource(path string) (*os.File, error) {
file, err := os.OpenFile(path, os.O_RDONLY|openNoAtime, 0)
if err == nil || !errors.Is(err, fs.ErrPermission) {
return file, err
}
return os.Open(path)
}
func findMountpoint(path string, mountPoints []string) string {
matched := ""
for _, mountPoint := range mountPoints {
if path != mountPoint {
prefix := mountPoint
if !strings.HasSuffix(prefix, string(filepath.Separator)) {
prefix += string(filepath.Separator)
}
if !strings.HasPrefix(path, prefix) {
continue
}
}
if len(mountPoint) > len(matched) {
matched = mountPoint
}
}
return matched
}