amadan

carlos / platform Public

stopevidence: writers-stopped allows only the replicator's identified descriptors, so a release can flush before full stop

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
Paul Campbell pushed by claude e13f69412ac7ae7edae0847f54c4dbdd79c82e29 parent cec0801
7 files changed, +919 −73
  • internal/stopevidence/fds_linux.go +78 −49
  • internal/stopevidence/reconstruct.go +26 −0
  • internal/stopevidence/reconstruct_linux.go +0 −24
  • internal/stopevidence/writers.go +155 −0
  • internal/stopevidence/writers_linux.go +259 −0
  • internal/stopevidence/writers_linux_test.go +386 −0
  • internal/stopevidence/writers_other.go +15 −0
diff --git a/internal/stopevidence/fds_linux.go b/internal/stopevidence/fds_linux.go
index 9cec70976..80f417f3c 100644
--- a/internal/stopevidence/fds_linux.go
+++ b/internal/stopevidence/fds_linux.go
@@ -31,29 +31,9 @@ func fdHolders(ctx context.Context, procRoot string, paths []string, ops fdOps)
if ctx == nil || !filepath.IsAbs(procRoot) || len(paths) == 0 {
return nil, ErrInvalidInput
}
- wanted := make(map[string]bool, len(paths))
- for _, path := range paths {
- if !filepath.IsAbs(path) {
- return nil, ErrInvalidInput
- }
- wanted[filepath.Clean(path)] = true
- }
- // The same file reached by another name (a hard link, another mount of
- // the file system) is the same inode: match on device and inode too.
- inodes := map[fileID]bool{}
- for path := range wanted {
- fi, err := ops.stat(path)
- if errors.Is(err, os.ErrNotExist) {
- continue
- }
- if err != nil {
- return nil, fmt.Errorf("%w: %s: %w", ErrIncomplete, path, err)
- }
- if id, ok := idOf(fi); ok {
- inodes[id] = true
- } else {
- return nil, fmt.Errorf("%w: %s: no inode", ErrIncomplete, path)
- }
+ files, err := newFileSet(paths, ops)
+ if err != nil {
+ return nil, err
}
if err := ctx.Err(); err != nil {
return nil, err
@@ -95,41 +75,19 @@ func fdHolders(ctx context.Context, procRoot string, paths []string, ops fdOps)
if err := ctx.Err(); err != nil {
return nil, err
}
- target, err := ops.readlink(filepath.Join(fdDir, fd.Name()))
+ fdPath := filepath.Join(fdDir, fd.Name())
+ target, err := ops.readlink(fdPath)
if procGone(err) {
continue
}
if err != nil {
return nil, fmt.Errorf("%w: pid %d tid %d fd %s: %w", ErrIncomplete, pid, tid, fd.Name(), err)
}
- if filepath.IsAbs(target) && (wanted[filepath.Clean(target)] || wanted[filepath.Clean(strings.TrimSuffix(target, " (deleted)"))]) {
- found = true
- continue
- }
- if len(inodes) == 0 || !filepath.IsAbs(target) {
- continue // sockets, pipes and anon inodes name no file
- }
- // Stat through the fd link reaches the open file itself, whatever
- // name it was opened by. An unreadable entry fails the whole scan.
- fdPath := filepath.Join(fdDir, fd.Name())
- fi, err := ops.stat(fdPath)
- if procGone(err) {
- // Closed only if the entry itself is gone. An entry that still
- // exists but cannot be followed is unreadable, not closed.
- if _, lerr := ops.lstat(fdPath); procGone(lerr) {
- continue
- } else if lerr == nil {
- err = fmt.Errorf("fd entry exists but its file cannot be reached: %w", err)
- } else {
- err = lerr
- }
- }
+ held, err := files.holds(fdPath, target, ops)
if err != nil {
return nil, fmt.Errorf("%w: pid %d tid %d fd %s: %w", ErrIncomplete, pid, tid, fd.Name(), err)
}
- if id, ok := idOf(fi); !ok {
- return nil, fmt.Errorf("%w: pid %d tid %d fd %s: no inode", ErrIncomplete, pid, tid, fd.Name())
- } else if inodes[id] {
+ if held {
found = true
}
}
@@ -165,6 +123,77 @@ func readProcDir(path, owner string, ops fdOps) ([]os.DirEntry, error) {
return entries, err
}
+// fileSet is the files a descriptor scan looks for, by path text and by
+// device and inode: the same file reached by another name (a hard link,
+// another mount of the file system) is the same inode.
+type fileSet struct {
+ paths map[string]bool
+ inodes map[fileID]bool
+}
+
+// newFileSet stats each of paths. A path that does not exist is matched by
+// its text alone; any other failure to stat it, or a file with no inode,
+// is ErrIncomplete. A relative path is ErrInvalidInput.
+func newFileSet(paths []string, ops fdOps) (fileSet, error) {
+ set := fileSet{paths: make(map[string]bool, len(paths)), inodes: map[fileID]bool{}}
+ for _, path := range paths {
+ if !filepath.IsAbs(path) {
+ return fileSet{}, ErrInvalidInput
+ }
+ set.paths[filepath.Clean(path)] = true
+ }
+ for path := range set.paths {
+ fi, err := ops.stat(path)
+ if errors.Is(err, os.ErrNotExist) {
+ continue
+ }
+ if err != nil {
+ return fileSet{}, fmt.Errorf("%w: %s: %w", ErrIncomplete, path, err)
+ }
+ if id, ok := idOf(fi); ok {
+ set.inodes[id] = true
+ } else {
+ return fileSet{}, fmt.Errorf("%w: %s: no inode", ErrIncomplete, path)
+ }
+ }
+ return set, nil
+}
+
+// holds reports whether the descriptor at fdPath, whose link text is
+// target, is one of the set's files. A descriptor that closed since its
+// link was read holds nothing. Any failure to tell is returned, for the
+// caller to fail its scan as incomplete: never a "no".
+func (s fileSet) holds(fdPath, target string, ops fdOps) (bool, error) {
+ if filepath.IsAbs(target) && (s.paths[filepath.Clean(target)] || s.paths[filepath.Clean(strings.TrimSuffix(target, " (deleted)"))]) {
+ return true, nil
+ }
+ if len(s.inodes) == 0 || !filepath.IsAbs(target) {
+ return false, nil // sockets, pipes and anon inodes name no file
+ }
+ // Stat through the fd link reaches the open file itself, whatever
+ // name it was opened by. An unreadable entry fails the whole scan.
+ fi, err := ops.stat(fdPath)
+ if procGone(err) {
+ // Closed only if the entry itself is gone. An entry that still
+ // exists but cannot be followed is unreadable, not closed.
+ if _, lerr := ops.lstat(fdPath); procGone(lerr) {
+ return false, nil
+ } else if lerr == nil {
+ err = fmt.Errorf("fd entry exists but its file cannot be reached: %w", err)
+ } else {
+ err = lerr
+ }
+ }
+ if err != nil {
+ return false, err
+ }
+ id, ok := idOf(fi)
+ if !ok {
+ return false, errors.New("no inode")
+ }
+ return s.inodes[id], nil
+}
+
type fileID struct{ dev, ino uint64 }
func idOf(fi os.FileInfo) (fileID, bool) {
diff --git a/internal/stopevidence/reconstruct.go b/internal/stopevidence/reconstruct.go
index 46466ac20..4a06368c1 100644
--- a/internal/stopevidence/reconstruct.go
+++ b/internal/stopevidence/reconstruct.go
@@ -2,6 +2,8 @@ package stopevidence
import (
"context"
+ "path/filepath"
+ "sort"
"github.com/carlosframework/platform/internal/cgroup"
)
@@ -23,3 +25,27 @@ type ReconstructInputs struct {
ProcRoot string
LeaseFree func(context.Context) (bool, error)
}
+
+// dbPathsWithSidecars is every database file with its -wal and -shm, clean,
+// sorted and without repeats. Each path must be absolute, and there must be
+// at least one: a cut always has a database path, so an empty list is a
+// caller's mistake, not "nothing to hold open".
+func dbPathsWithSidecars(paths []string) ([]string, error) {
+ if len(paths) == 0 {
+ return nil, ErrInvalidInput
+ }
+ seen := map[string]bool{}
+ for _, p := range paths {
+ if !filepath.IsAbs(p) {
+ return nil, ErrInvalidInput
+ }
+ p = filepath.Clean(p)
+ seen[p], seen[p+"-wal"], seen[p+"-shm"] = true, true, true
+ }
+ out := make([]string, 0, len(seen))
+ for p := range seen {
+ out = append(out, p)
+ }
+ sort.Strings(out)
+ return out, nil
+}
diff --git a/internal/stopevidence/reconstruct_linux.go b/internal/stopevidence/reconstruct_linux.go
index be40b6751..a396bdfe8 100644
--- a/internal/stopevidence/reconstruct_linux.go
+++ b/internal/stopevidence/reconstruct_linux.go
@@ -237,30 +237,6 @@ func strays(self int, reads ...[]int) []int {
return out
}
-// dbPathsWithSidecars is every database file with its -wal and -shm, clean,
-// sorted and without repeats. Each path must be absolute, and there must be
-// at least one: a cut always has a database path, so an empty list is a
-// caller's mistake, not "nothing to hold open".
-func dbPathsWithSidecars(paths []string) ([]string, error) {
- if len(paths) == 0 {
- return nil, ErrInvalidInput
- }
- seen := map[string]bool{}
- for _, p := range paths {
- if !filepath.IsAbs(p) {
- return nil, ErrInvalidInput
- }
- p = filepath.Clean(p)
- seen[p], seen[p+"-wal"], seen[p+"-shm"] = true, true, true
- }
- out := make([]string, 0, len(seen))
- for p := range seen {
- out = append(out, p)
- }
- sort.Strings(out)
- return out, nil
-}
-
// ownLeafProcs reads the pids in the agent's own cgroup leaf: the v2 path
// <procRoot>/self/cgroup names, under the cgroup2 mount
// <procRoot>/self/mountinfo names (ruling G32: never a hard-coded
diff --git a/internal/stopevidence/writers.go b/internal/stopevidence/writers.go
new file mode 100644
index 000000000..a162c8462
--- /dev/null
+++ b/internal/stopevidence/writers.go
@@ -0,0 +1,155 @@
+package stopevidence
+
+import (
+ "context"
+ "errors"
+ "path/filepath"
+ "time"
+
+ "github.com/carlosframework/platform/internal/controlname"
+)
+
+var (
+ ErrWorkOpen = errors.New("stopevidence: admission, spawns or work not drained")
+ ErrForeignHolder = errors.New("stopevidence: a descriptor other than the replicator's holds the database")
+)
+
+// Descriptor is one open file descriptor of one process, as /proc shows it.
+type Descriptor struct {
+ PID int
+ FD int
+ Target string
+}
+
+// WritersInputs is claim §6.1's writers-stopped observation (L10, region
+// HA spec 1 §5.8 release step 1). The caller's drain acknowledgement
+// supplies the three drained facts; this package checks the kernel.
+type WritersInputs struct {
+ Host string
+ // FenceClosed, SpawnsSettled and WorkDrained are the release
+ // controller's durable admission fence, settled spawn reservations, and
+ // drained application and timer work.
+ FenceClosed, SpawnsSettled, WorkDrained bool
+ Cgroup interface {
+ Populated() (bool, error)
+ Members() ([]int, error)
+ }
+ DBPaths []string
+ // Replicator names the only descriptors that may hold the database:
+ // the agent's own replicator for this host, by PID, its process start
+ // time (field 22 of /proc/<pid>/stat), and each descriptor.
+ ReplicatorPID int
+ ReplicatorStart string
+ Replicator []Descriptor
+ ProcRoot string
+}
+
+// WritersEvidence records what WritersStopped observed.
+type WritersEvidence struct {
+ Host string
+ CheckedAt time.Time
+ CheckedPaths []string
+ Held []Descriptor // the replicator descriptors still open
+ ProcRoot string
+}
+
+// WritersStopped passes only when admission is fenced, spawns settled,
+// work drained, the host's cgroup empty with no members, and every holder
+// of the database, WAL or SHM files is one of the replicator's identified
+// descriptors in the replicator's own process, started at
+// ReplicatorStart. An incomplete scan, an unreadable start time, or any
+// other holder fails. It grants no handover: full stop evidence (Collect)
+// still follows, after the replicator closes.
+func WritersStopped(ctx context.Context, in WritersInputs) (WritersEvidence, error) {
+ if ctx == nil || !controlname.ValidHostShape(in.Host) || len(in.DBPaths) == 0 || in.ReplicatorPID <= 0 || in.ReplicatorStart == "" {
+ return WritersEvidence{}, ErrInvalidInput
+ }
+ if !in.FenceClosed || !in.SpawnsSettled || !in.WorkDrained {
+ return WritersEvidence{}, ErrWorkOpen
+ }
+ if in.Cgroup == nil {
+ return WritersEvidence{}, ErrNoCgroupEvidence
+ }
+ paths, err := dbPathsWithSidecars(in.DBPaths)
+ if err != nil {
+ return WritersEvidence{}, err
+ }
+ if populated, err := in.Cgroup.Populated(); err != nil {
+ return WritersEvidence{}, err
+ } else if populated {
+ return WritersEvidence{}, ErrPopulated
+ }
+ if members, err := in.Cgroup.Members(); err != nil {
+ return WritersEvidence{}, err
+ } else if len(members) != 0 {
+ return WritersEvidence{}, ErrMembers
+ }
+ root := in.ProcRoot
+ if root == "" {
+ root = "/proc"
+ }
+ held, err := scanDescriptors(ctx, root, paths)
+ if err != nil {
+ return WritersEvidence{}, err
+ }
+ allowed := map[Descriptor]bool{}
+ for _, d := range in.Replicator {
+ if d.PID != in.ReplicatorPID {
+ return WritersEvidence{}, ErrInvalidInput
+ }
+ allowed[Descriptor{PID: d.PID, FD: d.FD, Target: filepath.Clean(d.Target)}] = true
+ }
+ for _, d := range held {
+ if !allowed[d] {
+ return WritersEvidence{}, ErrForeignHolder
+ }
+ }
+ if len(held) > 0 {
+ start, err := processStart(root, in.ReplicatorPID)
+ if err != nil {
+ return WritersEvidence{}, errors.Join(ErrIncomplete, err)
+ }
+ if start != in.ReplicatorStart {
+ return WritersEvidence{}, ErrForeignHolder
+ }
+ }
+ if err := ctx.Err(); err != nil {
+ return WritersEvidence{}, err
+ }
+ return WritersEvidence{Host: in.Host, CheckedAt: time.Now().UTC(), CheckedPaths: paths, Held: held, ProcRoot: root}, nil
+}
+
+// ProcessDescriptors lists the descriptors of one process (its leader's
+// table) that hold one of dbs' database, WAL or SHM files, as /proc shows
+// them. It reads only /proc/<pid>, never another process's table, so the
+// agent (uid carlos, no CAP_SYS_PTRACE) can take it for its own pid
+// immediately before and after it opens a host's replicator: the
+// difference is that replicator's descriptor set, which
+// WritersInputs.Replicator names at release (L10: "descriptors positively
+// identified as belonging to this host's admitted replicator"). An
+// unreadable table is ErrIncomplete, never an empty set; an empty dbs is
+// ErrInvalidInput. The complete scan of every process (WritersStopped)
+// stays root's.
+func ProcessDescriptors(ctx context.Context, procRoot string, pid int, dbs []string) ([]Descriptor, error) {
+ if ctx == nil || pid <= 0 {
+ return nil, ErrInvalidInput
+ }
+ paths, err := dbPathsWithSidecars(dbs)
+ if err != nil {
+ return nil, err
+ }
+ if procRoot == "" {
+ procRoot = "/proc"
+ }
+ held, err := scanProcess(ctx, procRoot, pid, paths)
+ if err != nil {
+ return nil, err
+ }
+ var own []Descriptor
+ for _, d := range held {
+ if d.FD >= 0 {
+ own = append(own, d)
+ }
+ }
+ return own, nil
+}
diff --git a/internal/stopevidence/writers_linux.go b/internal/stopevidence/writers_linux.go
new file mode 100644
index 000000000..c59068794
--- /dev/null
+++ b/internal/stopevidence/writers_linux.go
@@ -0,0 +1,259 @@
+//go:build linux
+
+package stopevidence
+
+import (
+ "context"
+ "errors"
+ "fmt"
+ "os"
+ "path/filepath"
+ "sort"
+ "strconv"
+ "strings"
+
+ "golang.org/x/sys/unix"
+)
+
+// kcmp(2)'s comparison types (linux/kcmp.h). golang.org/x/sys has the
+// syscall number (unix.SYS_KCMP) on every linux architecture but no
+// wrapper or constants.
+const (
+ kcmpFile = 0 // the same open file description: idx1, idx2 are fds
+ kcmpFiles = 2 // the same file descriptor table
+)
+
+// kcmpFunc is kcmp(2): 0 when the two tasks' resources are the same, a
+// positive ordering (1, 2) or 3 (not comparable) when they are not.
+type kcmpFunc func(pid1, pid2, typ int, idx1, idx2 uintptr) (int, error)
+
+func sysKcmp(pid1, pid2, typ int, idx1, idx2 uintptr) (int, error) {
+ r, _, errno := unix.Syscall6(unix.SYS_KCMP, uintptr(pid1), uintptr(pid2), uintptr(typ), idx1, idx2, 0)
+ if errno != 0 {
+ return 0, errno
+ }
+ return int(r), nil
+}
+
+// kcmpFor is the kernel's identity check for the tasks under root.
+// kcmp(2) takes pids in the caller's own pid namespace, so it is used
+// only for the caller's /proc; any other root (a test fixture) has none,
+// which fails closed as ErrIncomplete. Tests replace it. kcmp needs
+// ptrace read access to both tasks: always granted within the caller's
+// own process (ProcessDescriptors), and to root's complete scan through
+// CAP_SYS_PTRACE; an EPERM (a Yama scope that forbids it), or ENOSYS (a
+// kernel built without CONFIG_KCMP), is an incomplete observation.
+var kcmpFor = func(root string) kcmpFunc {
+ if filepath.Clean(root) == "/proc" {
+ return sysKcmp
+ }
+ return func(int, int, int, uintptr, uintptr) (int, error) { return 0, unix.ENOSYS }
+}
+
+// scanDescriptors lists every descriptor under root that is one of paths,
+// by path text or by device and inode (FDHolders' fileSet), deduplicated by
+// process and descriptor number. A descriptor
+// seen only in a task that does not share the leader's table (an unshared
+// thread) is reported against the leader's PID with FD -1, so no
+// allow-list can exempt it.
+func scanDescriptors(ctx context.Context, root string, paths []string) ([]Descriptor, error) {
+ ops := realFDOps()
+ if !filepath.IsAbs(root) {
+ return nil, ErrInvalidInput
+ }
+ files, err := newFileSet(paths, ops)
+ if err != nil {
+ return nil, err
+ }
+ entries, err := ops.readDir(root)
+ if err != nil {
+ return nil, fmt.Errorf("%w: %w", ErrIncomplete, err)
+ }
+ var out []Descriptor
+ for _, entry := range entries {
+ if err := ctx.Err(); err != nil {
+ return nil, err
+ }
+ pid, err := strconv.Atoi(entry.Name())
+ if err != nil || pid <= 0 {
+ continue
+ }
+ held, err := scanOne(ctx, root, pid, files, ops, kcmpFor(root))
+ if err != nil {
+ return nil, err
+ }
+ out = append(out, held...)
+ }
+ sort.Slice(out, func(i, j int) bool {
+ if out[i].PID != out[j].PID {
+ return out[i].PID < out[j].PID
+ }
+ return out[i].FD < out[j].FD
+ })
+ return out, nil
+}
+
+// scanProcess is scanDescriptors for one process: it reads only
+// /proc/<pid>, so it needs no privilege over any other process.
+func scanProcess(ctx context.Context, root string, pid int, paths []string) ([]Descriptor, error) {
+ if !filepath.IsAbs(root) {
+ return nil, ErrInvalidInput
+ }
+ ops := realFDOps()
+ files, err := newFileSet(paths, ops)
+ if err != nil {
+ return nil, err
+ }
+ if _, err := ops.stat(filepath.Join(root, strconv.Itoa(pid))); err != nil {
+ // A process asking about itself cannot have vanished: an absent
+ // or unreadable /proc/<pid> is an incomplete observation.
+ return nil, fmt.Errorf("%w: pid %d: %w", ErrIncomplete, pid, err)
+ }
+ out, err := scanOne(ctx, root, pid, files, ops, kcmpFor(root))
+ if err != nil {
+ return nil, err
+ }
+ sort.Slice(out, func(i, j int) bool { return out[i].FD < out[j].FD })
+ return out, nil
+}
+
+// scanOne lists pid's descriptors on files, deduplicated by descriptor
+// number. A thread's descriptor counts as the leader's only on positive
+// kernel evidence (round 6 finding 1): the leader shows the same number
+// for the same path, kcmp(2) KCMP_FILES says the thread shares the
+// leader's descriptor table, and KCMP_FILE says that descriptor is the
+// leader's open file description. A number and a path alone prove
+// nothing: a thread that unshared its table (CLONE_FILES) can reopen the
+// database at the same number. Any other thread's descriptor is reported
+// with FD -1, so no allow-list can exempt it. A comparison that cannot be
+// made (ENOSYS, EPERM, anything but a vanished task) is ErrIncomplete. A
+// process or thread that vanished mid-scan is skipped; an unreadable
+// table is ErrIncomplete.
+func scanOne(ctx context.Context, root string, pid int, files fileSet, ops fdOps, cmp kcmpFunc) ([]Descriptor, error) {
+ pidDir := filepath.Join(root, strconv.Itoa(pid))
+ tasks, err := readProcDir(filepath.Join(pidDir, "task"), pidDir, ops)
+ if err != nil {
+ return nil, fmt.Errorf("%w: pid %d: %w", ErrIncomplete, pid, err)
+ }
+ // The leader's table first: a thread's descriptor is the leader's
+ // only when the leader shows the same number for the same file.
+ leaderName := strconv.Itoa(pid)
+ sort.SliceStable(tasks, func(i, j int) bool { return tasks[i].Name() == leaderName && tasks[j].Name() != leaderName })
+ leader := map[int]string{}
+ seen := map[Descriptor]bool{}
+ var out []Descriptor
+ for _, task := range tasks {
+ if err := ctx.Err(); err != nil {
+ return nil, err
+ }
+ tid, err := strconv.Atoi(task.Name())
+ if err != nil || tid <= 0 {
+ continue
+ }
+ taskDir := filepath.Join(pidDir, "task", task.Name())
+ fdDir := filepath.Join(taskDir, "fd")
+ fds, err := readProcDir(fdDir, taskDir, ops)
+ if err != nil {
+ return nil, fmt.Errorf("%w: pid %d tid %d: %w", ErrIncomplete, pid, tid, err)
+ }
+ shared := -1 // this thread's KCMP_FILES verdict, asked once, only if needed
+ descriptors:
+ for _, fd := range fds {
+ n, err := strconv.Atoi(fd.Name())
+ if err != nil {
+ continue
+ }
+ fdPath := filepath.Join(fdDir, fd.Name())
+ target, err := ops.readlink(fdPath)
+ if procGone(err) {
+ continue
+ }
+ if err != nil {
+ return nil, fmt.Errorf("%w: pid %d tid %d fd %d: %w", ErrIncomplete, pid, tid, n, err)
+ }
+ // FDHolders' identity: the path text, or the device and inode
+ // through the link, so a hard link or another mount's path to
+ // the database is a holder too.
+ if held, err := files.holds(fdPath, target, ops); err != nil {
+ return nil, fmt.Errorf("%w: pid %d tid %d fd %d: %w", ErrIncomplete, pid, tid, n, err)
+ } else if !held {
+ continue
+ }
+ clean := filepath.Clean(strings.TrimSuffix(target, " (deleted)"))
+ d := Descriptor{PID: pid, FD: n, Target: clean}
+ if tid == pid {
+ leader[n] = clean
+ } else {
+ ok, gone, err := leadersDescriptor(cmp, pid, tid, n, leader[n] == clean, &shared)
+ if err != nil {
+ return nil, fmt.Errorf("%w: pid %d tid %d fd %d: kcmp: %w", ErrIncomplete, pid, tid, n, err)
+ }
+ if gone {
+ break descriptors // the thread exited mid-scan
+ }
+ if !ok {
+ d.FD = -1 // not provably the leader's descriptor
+ }
+ }
+ if !seen[d] {
+ seen[d] = true
+ out = append(out, d)
+ }
+ }
+ }
+ return out, nil
+}
+
+// leadersDescriptor is the kernel's evidence that thread tid's descriptor
+// n is the leader's own: the same table (KCMP_FILES, asked once per thread
+// and kept in *shared) and the same open file description (KCMP_FILE).
+// sameName is the leader showing n on the same path. gone reports a task
+// that exited (ESRCH); EBADF (n closed in one of them since its listing)
+// is "not the leader's"; any other error is returned.
+func leadersDescriptor(cmp kcmpFunc, pid, tid, n int, sameName bool, shared *int) (ok, gone bool, err error) {
+ if !sameName {
+ return false, false, nil
+ }
+ if *shared < 0 {
+ r, err := cmp(pid, tid, kcmpFiles, 0, 0)
+ if errors.Is(err, unix.ESRCH) {
+ return false, true, nil
+ }
+ if err != nil {
+ return false, false, err
+ }
+ *shared = r
+ }
+ if *shared != 0 {
+ return false, false, nil
+ }
+ r, err := cmp(pid, tid, kcmpFile, uintptr(n), uintptr(n))
+ switch {
+ case errors.Is(err, unix.ESRCH):
+ return false, true, nil
+ case errors.Is(err, unix.EBADF):
+ return false, false, nil
+ case err != nil:
+ return false, false, err
+ }
+ return r == 0, false, nil
+}
+
+// processStart is field 22 of /proc/<pid>/stat: the process start time in
+// clock ticks since boot, which with the boot ID names one process.
+func processStart(root string, pid int) (string, error) {
+ raw, err := os.ReadFile(filepath.Join(root, strconv.Itoa(pid), "stat"))
+ if err != nil {
+ return "", err
+ }
+ s := string(raw)
+ end := strings.LastIndexByte(s, ')')
+ if end < 0 {
+ return "", ErrIncomplete
+ }
+ fields := strings.Fields(s[end+1:])
+ if len(fields) < 20 { // fields 3.. start after the comm
+ return "", ErrIncomplete
+ }
+ return fields[19], nil
+}
diff --git a/internal/stopevidence/writers_linux_test.go b/internal/stopevidence/writers_linux_test.go
new file mode 100644
index 000000000..eb242af4b
--- /dev/null
+++ b/internal/stopevidence/writers_linux_test.go
@@ -0,0 +1,386 @@
+//go:build linux
+
+package stopevidence
+
+import (
+ "errors"
+ "os"
+ "os/exec"
+ "path/filepath"
+ "runtime"
+ "slices"
+ "strconv"
+ "strings"
+ "testing"
+
+ "golang.org/x/sys/unix"
+)
+
+// fixtureUnshared names the fixture threads that do not share their
+// leader's descriptor table; fixtureKcmpErr makes every comparison fail.
+var (
+ fixtureUnshared map[int]bool
+ fixtureKcmpErr error
+)
+
+// fixtureKcmp answers kcmp(2) for a /proc fixture as the kernel would: a
+// thread shares the leader's table unless fixtureUnshared names it, and a
+// shared table's descriptor n is the leader's n; a descriptor missing from
+// either table is EBADF.
+func fixtureKcmp(root string) kcmpFunc {
+ return func(pid, tid, typ int, idx1, idx2 uintptr) (int, error) {
+ if fixtureKcmpErr != nil {
+ return 0, fixtureKcmpErr
+ }
+ switch typ {
+ case kcmpFiles:
+ if fixtureUnshared[tid] {
+ return 1, nil
+ }
+ return 0, nil
+ case kcmpFile:
+ fd := func(task int, n uintptr) (string, error) {
+ return os.Readlink(filepath.Join(root, strconv.Itoa(pid), "task", strconv.Itoa(task), "fd", strconv.Itoa(int(n))))
+ }
+ a, errA := fd(pid, idx1)
+ b, errB := fd(tid, idx2)
+ if errA != nil || errB != nil {
+ return 0, unix.EBADF
+ }
+ if fixtureUnshared[tid] || a != b {
+ return 1, nil
+ }
+ return 0, nil
+ }
+ return 0, unix.EINVAL
+ }
+}
+
+// writersProc builds a /proc fixture: the agent (pid 20, threads 20 and
+// 7, sharing the table) holds the database on fds 5 and 6.
+func writersProc(t *testing.T) (string, string) {
+ t.Helper()
+ root, db := t.TempDir(), filepath.Join(t.TempDir(), "data.db")
+ fixtureUnshared, fixtureKcmpErr = map[int]bool{}, nil
+ prior := kcmpFor
+ kcmpFor = func(r string) kcmpFunc {
+ if r == root {
+ return fixtureKcmp(root)
+ }
+ return prior(r)
+ }
+ t.Cleanup(func() { kcmpFor = prior })
+ for _, tid := range []string{"20", "7"} {
+ dir := filepath.Join(root, "20", "task", tid, "fd")
+ if err := os.MkdirAll(dir, 0700); err != nil {
+ t.Fatal(err)
+ }
+ for fd, target := range map[int]string{5: db, 6: db + "-wal", 9: "/dev/null"} {
+ if err := os.Symlink(target, filepath.Join(dir, strconv.Itoa(fd))); err != nil {
+ t.Fatal(err)
+ }
+ }
+ }
+ if err := os.WriteFile(filepath.Join(root, "20", "stat"), []byte("20 (carlos edge) S 1 20 20 0 -1 4194560 1 0 0 0 0 0 0 0 20 0 9 0 777 0 0"), 0600); err != nil {
+ t.Fatal(err)
+ }
+ return root, db
+}
+
+func writersInputs(root, db string) WritersInputs {
+ return WritersInputs{Host: "shop.example", FenceClosed: true, SpawnsSettled: true, WorkDrained: true, Cgroup: &evidenceGroup{},
+ DBPaths: []string{db}, ReplicatorPID: 20, ReplicatorStart: "777", ProcRoot: root,
+ Replicator: []Descriptor{{PID: 20, FD: 5, Target: db}, {PID: 20, FD: 6, Target: db + "-wal"}}}
+}
+
+func TestWritersStoppedAllowsOnlyTheReplicatorsDescriptors(t *testing.T) {
+ root, db := writersProc(t)
+ ev, err := WritersStopped(t.Context(), writersInputs(root, db))
+ if err != nil || len(ev.Held) != 2 {
+ t.Fatalf("the replicator alone: %+v %v", ev, err)
+ }
+ for _, tc := range []struct {
+ name string
+ change func(*WritersInputs)
+ want error
+ }{
+ {"admission open", func(in *WritersInputs) { in.FenceClosed = false }, ErrWorkOpen},
+ {"spawns not settled", func(in *WritersInputs) { in.SpawnsSettled = false }, ErrWorkOpen},
+ {"timer work running", func(in *WritersInputs) { in.WorkDrained = false }, ErrWorkOpen},
+ {"cgroup populated", func(in *WritersInputs) { in.Cgroup = &evidenceGroup{populated: true} }, ErrPopulated},
+ {"cgroup members", func(in *WritersInputs) { in.Cgroup = &evidenceGroup{members: []int{8}} }, ErrMembers},
+ {"one replicator descriptor not identified", func(in *WritersInputs) { in.Replicator = in.Replicator[:1] }, ErrForeignHolder},
+ {"another process start", func(in *WritersInputs) { in.ReplicatorStart = "778" }, ErrForeignHolder},
+ {"descriptor of another process", func(in *WritersInputs) { in.Replicator[0].PID = 21 }, ErrInvalidInput},
+ } {
+ t.Run(tc.name, func(t *testing.T) {
+ in := writersInputs(root, db)
+ tc.change(&in)
+ if _, err := WritersStopped(t.Context(), in); !errors.Is(err, tc.want) {
+ t.Fatalf("got %v, want %v", err, tc.want)
+ }
+ })
+ }
+}
+
+func TestProcessDescriptorsNamesOneProcessesHolders(t *testing.T) {
+ root, db := writersProc(t)
+ // Another process holds the database too; it is not pid 20's.
+ dir := filepath.Join(root, "30", "task", "30", "fd")
+ if err := os.MkdirAll(dir, 0700); err != nil {
+ t.Fatal(err)
+ }
+ if err := os.Symlink(db, filepath.Join(dir, "4")); err != nil {
+ t.Fatal(err)
+ }
+ own, err := ProcessDescriptors(t.Context(), root, 20, []string{db})
+ if err != nil || len(own) != 2 || own[0] != (Descriptor{PID: 20, FD: 5, Target: db}) || own[1] != (Descriptor{PID: 20, FD: 6, Target: db + "-wal"}) {
+ t.Fatalf("own: %+v %v", own, err)
+ }
+ if _, err := ProcessDescriptors(t.Context(), filepath.Join(root, "missing"), 20, []string{db}); !errors.Is(err, ErrIncomplete) {
+ t.Fatalf("an unreadable table: %v", err)
+ }
+ // Another process's table being unreadable (root-owned, as the agent
+ // sees it) does not matter: only /proc/<pid> is read.
+ if err := os.Chmod(filepath.Join(root, "30", "task"), 0); err != nil {
+ t.Fatal(err)
+ }
+ t.Cleanup(func() { _ = os.Chmod(filepath.Join(root, "30", "task"), 0o700) })
+ if again, err := ProcessDescriptors(t.Context(), root, 20, []string{db}); err != nil || len(again) != 2 {
+ t.Fatalf("own descriptors with another table unreadable: %+v %v", again, err)
+ }
+ if _, err := WritersStopped(t.Context(), writersInputs(root, db)); !errors.Is(err, ErrIncomplete) {
+ t.Fatalf("the complete scan with an unreadable table must fail closed: %v", err)
+ }
+ if err := os.Chmod(filepath.Join(root, "30", "task"), 0o700); err != nil {
+ t.Fatal(err)
+ }
+ // The difference around an open is the replicator's set, and
+ // WritersStopped accepts exactly it.
+ in := writersInputs(root, db)
+ in.Replicator = own
+ if _, err := WritersStopped(t.Context(), in); !errors.Is(err, ErrForeignHolder) {
+ t.Fatalf("pid 30 still holds the database: %v", err)
+ }
+ if err := os.RemoveAll(filepath.Join(root, "30")); err != nil {
+ t.Fatal(err)
+ }
+ if _, err := WritersStopped(t.Context(), in); err != nil {
+ t.Fatal(err)
+ }
+}
+
+func TestWritersStoppedRefusesAnyOtherHolder(t *testing.T) {
+ for _, tc := range []struct{ name, pid, tid string }{
+ {"another process", "30", "30"},
+ {"an unshared thread of the agent", "20", "8"},
+ } {
+ t.Run(tc.name, func(t *testing.T) {
+ root, db := writersProc(t)
+ dir := filepath.Join(root, tc.pid, "task", tc.tid, "fd")
+ if err := os.MkdirAll(dir, 0700); err != nil {
+ t.Fatal(err)
+ }
+ if err := os.Symlink(db+"-shm", filepath.Join(dir, "11")); err != nil {
+ t.Fatal(err)
+ }
+ if _, err := WritersStopped(t.Context(), writersInputs(root, db)); !errors.Is(err, ErrForeignHolder) {
+ t.Fatalf("got %v", err)
+ }
+ })
+ }
+ t.Run("a thread's descriptor the leader's table does not show", func(t *testing.T) {
+ root, db := writersProc(t)
+ if err := os.Remove(filepath.Join(root, "20", "task", "20", "fd", "5")); err != nil {
+ t.Fatal(err)
+ }
+ if _, err := WritersStopped(t.Context(), writersInputs(root, db)); !errors.Is(err, ErrForeignHolder) {
+ t.Fatalf("got %v", err)
+ }
+ })
+ // Round 6 finding 1: a thread holding the database at the leader's
+ // number and path is the leader's only on the kernel's word.
+ t.Run("an unshared thread at the leader's number and path", func(t *testing.T) {
+ root, db := writersProc(t)
+ fixtureUnshared[7] = true
+ if _, err := WritersStopped(t.Context(), writersInputs(root, db)); !errors.Is(err, ErrForeignHolder) {
+ t.Fatalf("got %v", err)
+ }
+ })
+ t.Run("a shared thread whose descriptor is another description", func(t *testing.T) {
+ // The table compared shared, then the descriptor did not: the
+ // thread unshared and reopened between the two comparisons.
+ root, db := writersProc(t)
+ prior := kcmpFor
+ kcmpFor = func(string) kcmpFunc {
+ return func(pid, tid, typ int, a, b uintptr) (int, error) {
+ if typ == kcmpFile {
+ return 2, nil
+ }
+ return 0, nil
+ }
+ }
+ defer func() { kcmpFor = prior }()
+ if _, err := WritersStopped(t.Context(), writersInputs(root, db)); !errors.Is(err, ErrForeignHolder) {
+ t.Fatalf("got %v", err)
+ }
+ })
+ for _, kerr := range []error{unix.ENOSYS, unix.EPERM} {
+ t.Run("kcmp unavailable: "+kerr.Error(), func(t *testing.T) {
+ root, db := writersProc(t)
+ fixtureKcmpErr = kerr
+ if _, err := WritersStopped(t.Context(), writersInputs(root, db)); !errors.Is(err, ErrIncomplete) {
+ t.Fatalf("got %v", err)
+ }
+ if _, err := ProcessDescriptors(t.Context(), root, 20, []string{db}); !errors.Is(err, ErrIncomplete) {
+ t.Fatalf("own descriptors: %v", err)
+ }
+ })
+ }
+ t.Run("an unreadable descriptor directory", func(t *testing.T) {
+ root, db := writersProc(t)
+ if _, err := WritersStopped(t.Context(), writersInputs(filepath.Join(root, "missing"), db)); !errors.Is(err, ErrIncomplete) {
+ t.Fatalf("got %v", err)
+ }
+ })
+}
+
+// threadWith runs act on a new OS thread that has unshared its descriptor
+// table (CLONE_FILES) when unshare is set, and keeps the thread alive
+// until the returned release is called. The thread is never unlocked, so
+// it exits with its goroutine and the runtime never reuses its table.
+func threadWith(t *testing.T, unshare bool, act func() error) (tid int, release func()) {
+ t.Helper()
+ type started struct {
+ tid int
+ err error
+ }
+ ready, done := make(chan started), make(chan struct{})
+ var run func()
+ run = func() {
+ runtime.LockOSThread()
+ if unix.Gettid() == os.Getpid() {
+ // The leader's own thread: unsharing it would move the
+ // leader's table, not a thread's. Hold it and start again on
+ // another thread.
+ go run()
+ <-done
+ runtime.UnlockOSThread()
+ return
+ }
+ if unshare {
+ if err := unix.Unshare(unix.CLONE_FILES); err != nil {
+ ready <- started{err: err}
+ return
+ }
+ }
+ if err := act(); err != nil {
+ ready <- started{err: err}
+ return
+ }
+ ready <- started{tid: unix.Gettid()}
+ <-done
+ }
+ go run()
+ s := <-ready
+ if s.err != nil {
+ t.Fatal(s.err)
+ }
+ return s.tid, func() { close(done) }
+}
+
+// The real kernel, on this test process: a thread
+// sharing the table adds nothing; a thread that unshared it holds the
+// database at the leader's number and path, either by the copied
+// description or by reopening it independently there, and is reported
+// with FD -1, never as the leader's.
+func TestProcessDescriptorsTellsSharedFromUnsharedThreadsByKernelIdentity(t *testing.T) {
+ // The kernel facts are this process's own threads, so the test runs in
+ // a fresh copy of the test binary: another test in the package
+ // (main's TestFDHoldersUnsharedThread) may have unshared the leader
+ // thread's table for good, and then no thread shares the leader's.
+ if os.Getenv("STOPEVIDENCE_KCMP_CHILD") != "1" {
+ cmd := exec.Command(os.Args[0], "-test.run=^TestProcessDescriptorsTellsSharedFromUnsharedThreadsByKernelIdentity$", "-test.count=1", "-test.v")
+ cmd.Env = append(os.Environ(), "STOPEVIDENCE_KCMP_CHILD=1")
+ out, err := cmd.CombinedOutput()
+ if err != nil || !strings.Contains(string(out), "--- PASS: TestProcessDescriptorsTellsSharedFromUnsharedThreadsByKernelIdentity") {
+ t.Fatalf("the fresh process: %v\n%s", err, out)
+ }
+ return
+ }
+ db := filepath.Join(t.TempDir(), "data.db")
+ f, err := os.OpenFile(db, os.O_RDWR|os.O_CREATE, 0o600)
+ if err != nil {
+ t.Fatal(err)
+ }
+ defer f.Close()
+ n, pid := int(f.Fd()), os.Getpid()
+ own := Descriptor{PID: pid, FD: n, Target: db}
+ scan := func() []Descriptor {
+ t.Helper()
+ // scanOne through the one-process scan: the complete scan root
+ // makes (scanDescriptors) reads each process the same way.
+ got, err := scanProcess(t.Context(), "/proc", pid, []string{db})
+ if err != nil {
+ t.Fatal(err)
+ }
+ return got
+ }
+ tid, release := threadWith(t, false, func() error { return nil })
+ if tid == pid {
+ t.Fatal("the helper thread is the leader")
+ }
+ if got := scan(); !slices.Equal(got, []Descriptor{own}) {
+ t.Fatalf("a shared thread: %+v", got)
+ }
+ release()
+ for _, tc := range []struct {
+ name string
+ act func() error
+ }{
+ {"the copied description", func() error { return nil }},
+ {"reopened independently at the same number", func() error {
+ fd, err := unix.Open(db, unix.O_RDWR|unix.O_CLOEXEC, 0)
+ if err != nil {
+ return err
+ }
+ defer unix.Close(fd)
+ return unix.Dup3(fd, n, unix.O_CLOEXEC)
+ }},
+ } {
+ t.Run(tc.name, func(t *testing.T) {
+ _, release := threadWith(t, true, tc.act)
+ defer release()
+ got := scan()
+ if !slices.Contains(got, own) || !slices.Contains(got, Descriptor{PID: pid, FD: -1, Target: db}) {
+ t.Fatalf("an unshared thread passed as the leader's: %+v", got)
+ }
+ })
+ }
+}
+
+// The scan's identity is FDHolders' (main 8e9ac1b8): a process holding the
+// database through a hard link holds it, whatever its link text says.
+func TestWritersStoppedRefusesAHardLinkedHolder(t *testing.T) {
+ root, db := writersProc(t)
+ if err := os.WriteFile(db, []byte("SQLite format 3\x00"), 0o600); err != nil {
+ t.Fatal(err)
+ }
+ if _, err := WritersStopped(t.Context(), writersInputs(root, db)); err != nil {
+ t.Fatalf("the replicator alone, the database on disk: %v", err)
+ }
+ alias := filepath.Join(filepath.Dir(db), "alias.db")
+ if err := os.Link(db, alias); err != nil {
+ t.Skip("no hard links here:", err)
+ }
+ dir := filepath.Join(root, "30", "task", "30", "fd")
+ if err := os.MkdirAll(dir, 0700); err != nil {
+ t.Fatal(err)
+ }
+ if err := os.Symlink(alias, filepath.Join(dir, "4")); err != nil {
+ t.Fatal(err)
+ }
+ if _, err := WritersStopped(t.Context(), writersInputs(root, db)); !errors.Is(err, ErrForeignHolder) {
+ t.Fatalf("a holder through a hard link: %v", err)
+ }
+}
diff --git a/internal/stopevidence/writers_other.go b/internal/stopevidence/writers_other.go
new file mode 100644
index 000000000..af4c35dcc
--- /dev/null
+++ b/internal/stopevidence/writers_other.go
@@ -0,0 +1,15 @@
+//go:build !linux
+
+package stopevidence
+
+import "context"
+
+func scanDescriptors(context.Context, string, []string) ([]Descriptor, error) {
+ return nil, ErrUnsupported
+}
+
+func scanProcess(context.Context, string, int, []string) ([]Descriptor, error) {
+ return nil, ErrUnsupported
+}
+
+func processStart(string, int) (string, error) { return "", ErrUnsupported }