updater: cap updater.log, sweep abandoned bundle downloads; peer-churn goroutine gate

.github/workflows/architecture.yml

@@ -53,4 +53,4 @@ jobs:
         run: go test -race -count=1 -timeout 10m -run TestConcurrentDialEncryptDecrypt ./tests/...
 
       - name: P4/P9 - goroutine reaping
-        run: go test ./tests -run "TestDaemonShutdown|TestStreamLifecycleGoroutines|TestPluginShutdown"
+        run: go test ./tests -run "TestDaemonShutdown|TestStreamLifecycleGoroutines|TestPluginShutdown|TestPeerChurnDoesNotLeakGoroutines"

cmd/updater/logcap.go

@@ -0,0 +1,77 @@
+// SPDX-License-Identifier: AGPL-3.0-or-later
+
+package main
+
+import (
+	"context"
+	"flag"
+	"os"
+	"path/filepath"
+	"time"
+
+	"github.com/pilot-protocol/pilotprotocol/internal/logcap"
+)
+
+// install.sh's launchd job points the updater's stdout and stderr at
+// ~/.pilot/updater.log (StandardOutPath/StandardErrorPath). launchd opens
+// that file and never rotates it, and nothing else does either: one
+// laptop's updater.log held every line since the June install (2,205
+// lines, 310 KB), and each failed GitHub check adds a ~400-byte line an
+// hour. The daemon caps its own daemon.log from the inside
+// (cmd/daemon/logrotation.go); the updater now does the same with the same
+// package, keeping gzipped <log>.pilot.N.gz generations. Under systemd the
+// updater logs to journald and this is a no-op: stderr is not a regular
+// file there.
+
+const (
+	// defaultUpdaterLogMaxMB is -log-max-size's default. The updater writes
+	// a few lines an hour, so this is years of history.
+	defaultUpdaterLogMaxMB = 10
+	// defaultUpdaterLogBackups is -log-max-backups's default, the same as
+	// the daemon's.
+	defaultUpdaterLogBackups = 3
+	// updaterLogCheckInterval is how often the log's size is checked.
+	updaterLogCheckInterval = 5 * time.Minute
+)
+
+// updaterLogDirs returns the directories the updater's log is in when
+// Pilot set it up, where the cap applies by default: ~/.pilot (install.sh's
+// launchd updater.log), and $PILOT_HOME/.pilot when that is set. A log an
+// operator redirected elsewhere may be rotated by other means, so it is
+// left alone unless -log-max-size is set explicitly.
+func updaterLogDirs() []string {
+	var dirs []string
+	if home, err := os.UserHomeDir(); err == nil {
+		dirs = append(dirs, filepath.Join(home, ".pilot"))
+	}
+	if h := os.Getenv("PILOT_HOME"); h != "" {
+		dirs = append(dirs, filepath.Join(h, ".pilot"))
+	}
+	return dirs
+}
+
+// capUpdaterLog starts capping the log f writes to (the updater's
+// stderr) at maxMB megabytes, keeping backups gzipped generations, until
+// ctx is done. anywhere lifts the ~/.pilot scope. It reports whether
+// capping started: false when maxMB <= 0, f is not a regular file, or the
+// log is out of scope.
+func capUpdaterLog(ctx context.Context, f *os.File, maxMB, backups int, anywhere bool, interval time.Duration) bool {
+	return logcap.Watch(ctx, f, logcap.Options{
+		MaxBytes:   int64(maxMB) << 20,
+		MaxBackups: backups,
+		Anywhere:   anywhere,
+		Within:     updaterLogDirs(),
+	}, interval)
+}
+
+// flagSetOnCommandLine reports whether flag name was given on the command
+// line.
+func flagSetOnCommandLine(name string) bool {
+	set := false
+	flag.Visit(func(f *flag.Flag) {
+		if f.Name == name {
+			set = true
+		}
+	})
+	return set
+}

cmd/updater/main.go

@@ -78,6 +78,8 @@ func main() {
 	// stays on. Mirrors the --state-path pattern with an env fallback.
 	skipAttestation := flag.Bool("skip-attestation", envBool("PILOT_UPDATER_SKIP_ATTESTATION"),
 		"skip SLSA attestation verification (default off); for test/air-gapped use only — production verifies in-process, no `gh` needed")
+	logMaxSize := flag.Int("log-max-size", defaultUpdaterLogMaxMB, "rotate the updater's log once it exceeds this many MB (copy-truncate into gzipped <log>.pilot.N.gz backups). By default only a log inside ~/.pilot is rotated (install.sh's launchd updater.log); a log elsewhere only when this is set explicitly. 0 disables")
+	logMaxBackups := flag.Int("log-max-backups", defaultUpdaterLogBackups, "gzipped generations kept by -log-max-size rotation (<log>.pilot.1.gz ... <log>.pilot.N.gz); 0 keeps none")
 	flag.Parse()
 
 	if *showVersion {
@@ -91,6 +93,10 @@ func main() {
 	}
 
 	setupLogging(*logLevel, *logFormat)
+	// launchd never rotates ~/.pilot/updater.log; cap it from the inside
+	// (logcap.go). Lives for the process's lifetime.
+	capUpdaterLog(context.Background(), os.Stderr, *logMaxSize, *logMaxBackups,
+		flagSetOnCommandLine("log-max-size"), updaterLogCheckInterval)
 
 	u := updater.New(updater.Config{
 		CheckInterval:   *interval,
@@ -124,6 +130,9 @@ func main() {
 	} else {
 		slog.Info("app auto-upgrade opted out (PILOT_APP_UPDATE_OPT_OUT); apps stay at their installed versions, pilot binaries still update")
 	}
+	// Remove abandoned bundle downloads from the temp dir (tmpsweep.go).
+	stopSweep := make(chan struct{})
+	go tempSweepLoop(*interval, stopSweep)
 	if *pin != "" {
 		slog.Info("version pinned", "tag", *pin)
 	}
@@ -136,6 +145,7 @@ func main() {
 	<-sig
 
 	slog.Info("shutting down")
+	close(stopSweep)
 	u.Stop()
 }
 

cmd/updater/owner_other.go

@@ -0,0 +1,11 @@
+// SPDX-License-Identifier: AGPL-3.0-or-later
+
+//go:build !unix
+
+package main
+
+import "os"
+
+// ownedByCurrentUser cannot tell the owner here, so the temp sweep removes
+// nothing.
+func ownedByCurrentUser(os.FileInfo) bool { return false }

cmd/updater/owner_unix.go

@@ -0,0 +1,17 @@
+// SPDX-License-Identifier: AGPL-3.0-or-later
+
+//go:build unix
+
+package main
+
+import (
+	"os"
+	"syscall"
+)
+
+// ownedByCurrentUser reports whether fi belongs to the user this process
+// runs as.
+func ownedByCurrentUser(fi os.FileInfo) bool {
+	st, ok := fi.Sys().(*syscall.Stat_t)
+	return ok && int(st.Uid) == os.Getuid()
+}

cmd/updater/tmpsweep.go

@@ -0,0 +1,144 @@
+// SPDX-License-Identifier: AGPL-3.0-or-later
+
+package main
+
+import (
+	"io/fs"
+	"log/slog"
+	"os"
+	"path/filepath"
+	"time"
+)
+
+// Stale download leftovers in the system temp directory.
+//
+// `pilotctl appstore install <catalogue id>`, and `appstore upgrade`, which
+// appUpgradeLoop runs every hour as `upgrade --all`, download each bundle to
+// $TMPDIR/pilot-bundle-<n>.tar.gz and unpack it into
+// $TMPDIR/pilot-bundle-unpack-<n>. The unpacked copy is not removed once the
+// install has consumed it, and the tarball survives a pilotctl that is
+// killed mid-download. The updater's own release download dir,
+// $TMPDIR/pilot-update-<n>, is removed when the update returns but not when
+// the process is killed mid-download. One laptop's $TMPDIR held 272 of
+// these. Nothing else removes them: macOS purges $TMPDIR only after days
+// without access, and a Linux /tmp that is not a tmpfs is often never
+// cleaned.
+//
+// sweepStaleTemps removes them once they are clearly abandoned. It only
+// touches entries that:
+//   - match one of these Pilot-specific names,
+//   - are the expected type (a directory for the two dir patterns, a
+//     regular file for the tarball; symlinks never match),
+//   - are owned by the user the updater runs as, and
+//   - were last modified more than staleTempAge ago. An install or update
+//     finishes in minutes (pilotctl's download timeout is 60 s, the
+//     updater's app-upgrade run is capped at 10 minutes), so a live one is
+//     never near that age.
+
+// staleTempAge is how old a leftover must be before it is removed.
+const staleTempAge = 24 * time.Hour
+
+// staleTempPattern is one name the sweep removes, and the type of entry it
+// must be.
+type staleTempPattern struct {
+	glob string
+	dir  bool
+}
+
+// staleTempPatterns are the temp names pilotctl (appstore_catalogue.go,
+// fetchAndUnpackBundle) and the updater library (applyUpdate) create with
+// os.CreateTemp/os.MkdirTemp in the system temp directory. Keep in step
+// with them.
+var staleTempPatterns = []staleTempPattern{
+	{glob: "pilot-bundle-unpack-*", dir: true},
+	{glob: "pilot-bundle-*.tar.gz", dir: false},
+	{glob: "pilot-update-*", dir: true},
+}
+
+// sweepResult summarises one sweepStaleTemps run.
+type sweepResult struct {
+	Removed int
+	Bytes   int64
+	Failed  int
+}
+
+// sweepStaleTemps removes the abandoned leftovers in dir (see above) last
+// modified before now-maxAge.
+func sweepStaleTemps(dir string, now time.Time, maxAge time.Duration) sweepResult {
+	var res sweepResult
+	cutoff := now.Add(-maxAge)
+	for _, p := range staleTempPatterns {
+		matches, err := filepath.Glob(filepath.Join(dir, p.glob))
+		if err != nil {
+			continue // only ErrBadPattern, and the patterns are constant
+		}
+		for _, path := range matches {
+			fi, err := os.Lstat(path)
+			if err != nil {
+				continue
+			}
+			if p.dir != fi.IsDir() || (!p.dir && !fi.Mode().IsRegular()) {
+				continue
+			}
+			if !fi.ModTime().Before(cutoff) || !ownedByCurrentUser(fi) {
+				continue
+			}
+			size := treeSize(path)
+			// os.RemoveAll does not follow symlinks inside the tree. Every
+			// directory in these trees is created owner-writable (0755/0700:
+			// pilotctl's untarUnder, the updater's staging), so nothing in
+			// them blocks removal.
+			if err := os.RemoveAll(path); err != nil {
+				res.Failed++
+				slog.Warn("temp sweep: could not remove stale download leftover", "path", path, "err", err)
+				continue
+			}
+			res.Removed++
+			res.Bytes += size
+		}
+	}
+	return res
+}
+
+// treeSize is the total size of the regular files under path (path itself
+// when it is a file). Best effort, for the log line only.
+func treeSize(path string) int64 {
+	var total int64
+	_ = filepath.WalkDir(path, func(_ string, d fs.DirEntry, err error) error {
+		if err != nil {
+			return nil
+		}
+		if d.Type().IsRegular() {
+			if info, err := d.Info(); err == nil {
+				total += info.Size()
+			}
+		}
+		return nil
+	})
+	return total
+}
+
+// tempSweepLoop runs sweepStaleTemps on the system temp directory now and
+// then every interval, until stop is closed. It runs whether or not
+// automatic updates are enabled: an interactive `pilotctl appstore install`
+// leaves the same leftovers.
+func tempSweepLoop(interval time.Duration, stop <-chan struct{}) {
+	sweep := func() {
+		res := sweepStaleTemps(os.TempDir(), time.Now(), staleTempAge)
+		if res.Removed > 0 || res.Failed > 0 {
+			slog.Info("temp sweep: removed stale download leftovers",
+				"dir", os.TempDir(), "removed", res.Removed, "bytes", res.Bytes, "failed", res.Failed)
+		}
+	}
+	sweep()
+	t := time.NewTicker(interval)
+	defer t.Stop()
+	for {
+		select {
+		case <-stop:
+			return
+		case <-t.C:
+			sweep()
+		}
+	}
+}