Fix datagram port leak, 40ms stream stalls and several pilotctl failure paths

CHANGELOG.md

@@ -228,6 +228,15 @@ Detailed per-release notes are on the
 - SIGHUP no longer reloads control state. It is still caught and ignored.
 
 ### Changed
+- **The tunnel socket asks the kernel for 4 MB buffers** in each direction
+  instead of the default (about 200 KB on Linux). Every tunnel shares the one
+  socket, and several streams sending at once overflowed it. The kernel caps
+  the request at `net.core.rmem_max` / `wmem_max`; raise those to get the full
+  size.
+- **The daemon's `info` reply is cheaper to build.** `pilotctl send-message`
+  asks for it on every send, and with a few hundred connections in the table
+  it was three quarters of the daemon's CPU under send load. The JSON is
+  unchanged.
 - **Dependencies: skillinject v0.2.4, dataexchange v0.2.3, updater v0.2.5**
   (plus common v0.5.14, and the sigstore-go v1.3.0 / go-openapi versions
   updater v0.2.5 requires).
@@ -251,6 +260,39 @@ Detailed per-release notes are on the
     `~/.pilot/update-state.json`.
 
 ### Fixed
+- **Datagrams no longer use up the daemon's ports.** Every datagram sent
+  (`pilotctl dgram`, `SendTo`, broadcasts) took an ephemeral source port and
+  never gave it back, because only closing a connection released one. After
+  about 16,000 datagrams no stream could be opened — sends failed with what
+  looked like an unreachable peer — until the daemon restarted. The port is now
+  released as soon as the datagram is sent. A dial that does fail for lack of
+  ports says so instead of telling the operator to check the peer.
+- **Large writes and first exchanges no longer stall for 40ms with both ends
+  idle.** A write held by Nagle waits for the data before it to be
+  acknowledged, and the receiver delayed a lone or odd segment's ACK by 40ms.
+  That cost 40ms on the first exchange of every connection, and one stall per
+  48KB chunk of a file transfer — about 1.5 MB/s on any link. The delayed-ACK
+  timer is now 5ms: a 60 MB transfer between two containers went from 31s to
+  3.5s.
+- **A dial returns when its handshake completes**, not on the next 10ms poll.
+  A local dial took 11ms and now takes under 1ms; with the change above,
+  connecting and exchanging one message dropped from 55ms to 6ms.
+- **`pilotctl send-message` fails when the receiver could not store the
+  message.** The receiver answers `ERR ...` when its disk is full or its inbox
+  is unwritable; the command printed that as the ack and exited 0. It now exits
+  non-zero, as `send-file` already did.
+- **`pilotctl appstore install` cleans up after itself.** Each install unpacked
+  the bundle into a temporary directory and left it there — a copy of every app
+  ever installed, and on a small `/tmp` the reason later installs failed with
+  "no space left on device". The directory is removed when the install ends,
+  whether it succeeded or not.
+- **`pilotctl appstore call` waits for an app that was just installed.** The
+  daemon starts an app a few seconds after `install` returns, so a call made
+  straight away failed with "socket not present". It now waits up to 15s for
+  the socket, and says so plainly when the app is suspended after crashing.
+- **`pilotctl received --clear` removes interrupted transfers too.** Their
+  bytes stay in `received/.partial` for a resume and count against the disk,
+  but were never listed or cleared. The reply gains a `cleared_partial` count.
 - **Parallel sends to one peer no longer stall or time out on the peer's
   SYN limiter.** The per-source limit admitted 10 connections at once and
   silently dropped the rest; the dialers retransmitted together, so a burst

cmd/pilotctl/appstore.go

@@ -1234,6 +1234,14 @@
 			"the argument must be either a catalogue ID (`pilotctl appstore catalogue` to list) or a path to a bundle dir containing manifest.json",
 			"%v", err)
 	}
+	if source == installSourceCatalogue {
+		// The bundle was unpacked into a temporary directory for this
+		// install alone. It was never removed, so every install — failed or
+		// not — left a copy of the app behind in $TMPDIR.
+		removeUnpacked := func() { _ = os.RemoveAll(bundleDir) }
+		defer removeUnpacked()
+		fatalCleanups = append(fatalCleanups, removeUnpacked)
+	}
 	if source == installSourceLocal && !allowLocal {
 		fatalHint("invalid_argument",
 			"local sideloads carry no catalogue signature; pass --local to confirm you trust this bundle's source. The supervisor will clamp the manifest to a small allow-list (fs.read/fs.write under $APP, audit.log). No net.dial, key.sign, ipc.call to other apps, or daemon hooks.",
@@ -2663,13 +2671,14 @@
 		}
 	}
 
-	sockPath := filepath.Join(appStoreRoot(), appID, "app.sock")
-	if _, err := os.Stat(sockPath); err != nil {
+	appDir := filepath.Join(appStoreRoot(), appID)
+	sockPath := filepath.Join(appDir, "app.sock")
+	if err := waitForAppSocket(appDir, sockPath, appSocketWait); err != nil {
 		// Not installed here at all: if the id is a rename tombstone, say so
 		// instead of blaming the daemon. No silent retarget: the method
 		// namespace changed too. (An installed app whose daemon is down skips
 		// the catalogue fetch.)
-		if _, derr := os.Stat(filepath.Dir(sockPath)); errors.Is(derr, os.ErrNotExist) { // #nosec G703 -- only a stat of the dir whose app.sock was stat'ed just above; nothing is read or written
+		if _, derr := os.Stat(appDir); errors.Is(derr, os.ErrNotExist) { // #nosec G703 -- only a stat of the dir whose app.sock was stat'ed just above; nothing is read or written
 			if c, lerr := loadCatalogue(); lerr == nil {
 				if e := c.findEntry(appID); e != nil && e.RenamedTo != "" {
 					fatalHint("invalid_argument",
@@ -2678,6 +2687,11 @@
 				}
 			}
 		}
+		if _, serr := os.Stat(filepath.Join(appDir, ".suspended")); serr == nil {
+			fatalHint("io_error",
+				"the app crashed repeatedly and was suspended: `pilotctl appstore audit "+appID+"` shows the exits, the app's own error output is in the daemon's log; `pilotctl appstore restart "+appID+"` retries it",
+				"app %s is suspended", appID)
+		}
 		fatalHint("io_error",
 			"is the daemon running and has it supervised this app yet?",
 			"socket %s not present: %v", sockPath, err)
@@ -2787,3 +2801,32 @@
 	_, _ = pretty.WriteTo(os.Stdout)
 	fmt.Println()
 }
+
+// appSocketWait is how long `appstore call` waits for an installed app's
+// socket to appear. The daemon starts an app on its next rescan after the
+// install and the adapter then stages its own assets, so a call made right
+// after `install` used to fail with "socket not present" for several seconds.
+const appSocketWait = 15 * time.Second
+
+// waitForAppSocket returns once sockPath exists. It only waits for an app
+// that is installed and not suspended; otherwise the socket is not coming and
+// it reports that at once.
+func waitForAppSocket(appDir, sockPath string, wait time.Duration) error {
+	deadline := time.Now().Add(wait)
+	for {
+		_, err := os.Stat(sockPath)
+		if err == nil {
+			return nil
+		}
+		if _, merr := os.Stat(filepath.Join(appDir, "manifest.json")); merr != nil {
+			return err // not installed
+		}
+		if _, serr := os.Stat(filepath.Join(appDir, ".suspended")); serr == nil {
+			return err
+		}
+		if time.Now().After(deadline) {
+			return err
+		}
+		time.Sleep(200 * time.Millisecond)
+	}
+}

cmd/pilotctl/appstore_catalogue.go

@@ -464,7 +464,8 @@ func resolveInstallTarget(target string) (string, installSource, error) {
 // fetchAndUnpackBundle downloads the catalogue entry's tarball,
 // verifies its sha256 against the catalogue value (defence against a
 // CDN substitute), and unpacks it into a tempdir whose path is
-// returned for the install path to consume.
+// returned for the install path to consume. The caller removes that
+// directory when it is done; on error nothing is left behind.
 func fetchAndUnpackBundle(e catalogueEntry) (string, error) {
 	bundleURL, bundleSHA, err := e.resolveBundle()
 	if err != nil {
@@ -518,15 +519,20 @@ func fetchAndUnpackBundle(e catalogueEntry) (string, error) {
 	}
 	f, err := os.Open(tmpTar.Name())
 	if err != nil {
+		_ = os.RemoveAll(unpackDir)
 		return "", err
 	}
 	defer f.Close()
 	gz, err := gzip.NewReader(f)
 	if err != nil {
+		_ = os.RemoveAll(unpackDir)
 		return "", fmt.Errorf("gzip: %w", err)
 	}
 	defer gz.Close()
 	if err := untarUnder(gz, unpackDir); err != nil {
+		// A partial unpack is what fills a small $TMPDIR: remove it, or
+		// the next install starts with less room than this one had.
+		_ = os.RemoveAll(unpackDir)
 		return "", fmt.Errorf("untar: %w", err)
 	}
 	return unpackDir, nil

cmd/pilotctl/fatal_trap.go

@@ -27,9 +27,17 @@ func exitAfterFatal(code, msg string) {
 	if fatalTrapDepth > 0 {
 		panic(trappedFatal{Code: code, Message: msg})
 	}
+	for _, fn := range fatalCleanups {
+		fn()
+	}
 	os.Exit(1)
 }
 
+// fatalCleanups run before a fatal error exits the process. os.Exit skips
+// deferred calls, so anything that must not outlive the command (a temporary
+// directory, say) registers here as well as deferring its own removal.
+var fatalCleanups []func()
+
 // runTrappingFatal runs fn and returns the fatal error it ended with, or nil
 // when it returned normally. Any other panic propagates.
 func runTrappingFatal(fn func()) (failure *trappedFatal) {

cmd/pilotctl/firstcontact.go

@@ -132,6 +132,10 @@ func dialFailureHint(target string, err error, attempts int, elapsed time.Durati
 		s = err.Error()
 	}
 	switch {
+	case strings.Contains(s, "ephemeral ports exhausted"):
+		// A local condition: the peer was never contacted, so pointing at
+		// its reachability sends the operator the wrong way.
+		return "this node has no free local ports for a new connection; close idle connections (`pilotctl connections`) or restart the daemon"
 	case strings.Contains(s, "key exchange"):
 		return fmt.Sprintf("%s never completed the tunnel key exchange (%d dial attempt(s), %s): the service is overloaded or offline. "+
 			"Retry in a minute; `pilotctl peers` shows the tunnel state.", target, attempts, elapsed.Round(time.Second))

cmd/pilotctl/main.go

@@ -4974,6 +4974,12 @@ func cmdSendMessage(args []string) {
 		defer cl.Close()
 		r := sendOne(cl, 0, false)
 		ackAt := time.Now()
+		// The receiver answers "ERR ..." when it could not store the
+		// message (disk full, inbox unwritable). That is a failed send, not
+		// a delivered one — same rule send-file applies.
+		if ackText, _ := r["ack"].(string); strings.HasPrefix(ackText, "ERR ") {
+			fatalCode("internal", "receiver rejected message: %s", ackText)
+		}
 		result := map[string]interface{}{
 			"target": target.String(),
 			"to":     target.String(),
@@ -6821,10 +6827,35 @@ func cmdReceived(args []string) {
 				count++
 			}
 		}
+		// Interrupted transfers leave their bytes in .partial for a resume.
+		// They count against the disk like any received file but were never
+		// listed or cleared, so a failed large transfer stayed forever.
+		partials := 0
+		partialDir := filepath.Join(dir, ".partial")
+		if entries, perr := os.ReadDir(partialDir); perr == nil {
+			for _, e := range entries {
+				info, ierr := e.Info()
+				if ierr != nil || e.IsDir() {
+					continue
+				}
+				if !sinceCutoff.IsZero() && info.ModTime().Before(sinceCutoff) {
+					continue
+				}
+				if !beforeCutoff.IsZero() && !info.ModTime().Before(beforeCutoff) {
+					continue
+				}
+				if os.Remove(filepath.Join(partialDir, e.Name())) == nil {
+					partials++
+				}
+			}
+		}
 		if jsonOutput {
-			outputOK(map[string]interface{}{"cleared": count, "remaining": len(all) - count})
+			outputOK(map[string]interface{}{"cleared": count, "remaining": len(all) - count, "cleared_partial": partials})
 		} else {
 			fmt.Printf("cleared %d received file(s), %d remaining\n", count, len(all)-count)
+			if partials > 0 {
+				fmt.Printf("cleared %d interrupted transfer(s)\n", partials)
+			}
 		}
 		return
 	}

cmd/pilotctl/zz_appstore_socket_wait_test.go

@@ -0,0 +1,68 @@
+// SPDX-License-Identifier: AGPL-3.0-or-later
+
+package main
+
+import (
+	"os"
+	"path/filepath"
+	"testing"
+	"time"
+)
+
+// `appstore call` right after `appstore install` used to fail: the daemon had
+// not started the app yet. It now waits for the socket — but only for an app
+// that is installed and not suspended, so a call that cannot succeed still
+// fails at once.
+func TestWaitForAppSocket(t *testing.T) {
+	t.Run("not installed fails at once", func(t *testing.T) {
+		dir := t.TempDir()
+		start := time.Now()
+		if err := waitForAppSocket(dir, filepath.Join(dir, "app.sock"), 5*time.Second); err == nil {
+			t.Fatal("want an error for an app with no manifest")
+		}
+		if time.Since(start) > time.Second {
+			t.Errorf("waited %v for an app that is not installed", time.Since(start))
+		}
+	})
+
+	t.Run("suspended fails at once", func(t *testing.T) {
+		dir := t.TempDir()
+		for _, name := range []string{"manifest.json", ".suspended"} {
+			if err := os.WriteFile(filepath.Join(dir, name), nil, 0o600); err != nil {
+				t.Fatal(err)
+			}
+		}
+		start := time.Now()
+		if err := waitForAppSocket(dir, filepath.Join(dir, "app.sock"), 5*time.Second); err == nil {
+			t.Fatal("want an error for a suspended app")
+		}
+		if time.Since(start) > time.Second {
+			t.Errorf("waited %v for a suspended app", time.Since(start))
+		}
+	})
+
+	t.Run("installed app is waited for", func(t *testing.T) {
+		dir := t.TempDir()
+		if err := os.WriteFile(filepath.Join(dir, "manifest.json"), nil, 0o600); err != nil {
+			t.Fatal(err)
+		}
+		sock := filepath.Join(dir, "app.sock")
+		go func() {
+			time.Sleep(400 * time.Millisecond)
+			_ = os.WriteFile(sock, nil, 0o600)
+		}()
+		if err := waitForAppSocket(dir, sock, 5*time.Second); err != nil {
+			t.Fatalf("socket appeared but wait failed: %v", err)
+		}
+	})
+
+	t.Run("gives up after the wait", func(t *testing.T) {
+		dir := t.TempDir()
+		if err := os.WriteFile(filepath.Join(dir, "manifest.json"), nil, 0o600); err != nil {
+			t.Fatal(err)
+		}
+		if err := waitForAppSocket(dir, filepath.Join(dir, "app.sock"), 300*time.Millisecond); err == nil {
+			t.Fatal("want an error when the socket never appears")
+		}
+	})
+}