diff --git a/cmd/wallet/lifecycle_test.go b/cmd/wallet/lifecycle_test.go new file mode 100644 index 0000000..3e1b646 --- /dev/null +++ b/cmd/wallet/lifecycle_test.go @@ -0,0 +1,252 @@ +//go:build unix + +package main + +import ( + "bufio" + "bytes" + "context" + "encoding/json" + "errors" + "fmt" + "io" + "log" + "net" + "os" + "os/exec" + "path/filepath" + "strconv" + "strings" + "sync" + "syscall" + "testing" + "time" + + "github.com/pilot-protocol/app-store/pkg/ipc" +) + +// The test binary doubles as the processes of the parent-death test: +// +// WALLET_TEST_ROLE=parent stands in for the pilot daemon: starts the +// wallet (own process group, no Pdeathsig — the +// supervisor's macOS setup), prints its pid, +// then blocks until it is killed. +// WALLET_TEST_ROLE=wallet runs the real main() with WALLET_TEST_ARGS. +func TestMain(m *testing.M) { + switch os.Getenv("WALLET_TEST_ROLE") { + case "parent": + os.Exit(fakeDaemon()) + case "wallet": + var args []string + if err := json.Unmarshal([]byte(os.Getenv("WALLET_TEST_ARGS")), &args); err != nil { + fmt.Fprintln(os.Stderr, "WALLET_TEST_ARGS:", err) + os.Exit(2) + } + os.Args = append([]string{os.Args[0]}, args...) + main() + os.Exit(0) + } + os.Exit(m.Run()) +} + +func fakeDaemon() int { + exe, err := os.Executable() + if err != nil { + fmt.Fprintln(os.Stderr, err) + return 2 + } + logf, err := os.Create(os.Getenv("WALLET_TEST_LOG")) + if err != nil { + fmt.Fprintln(os.Stderr, err) + return 2 + } + cmd := exec.Command(exe, "-test.run=^$") + cmd.Env = append(os.Environ(), "WALLET_TEST_ROLE=wallet") + cmd.Stdout, cmd.Stderr = logf, logf + cmd.SysProcAttr = &syscall.SysProcAttr{Setpgid: true} + if err := cmd.Start(); err != nil { + fmt.Fprintln(os.Stderr, err) + return 2 + } + fmt.Println(cmd.Process.Pid) + // Stay alive until the test SIGKILLs us, like a daemon that dies hard. + // (A bare select{} would trip the runtime's deadlock detector.) + time.Sleep(time.Hour) + return 0 +} + +// processGone reports whether pid has exited. A zombie counts as gone: in +// a container whose pid 1 does not reap, the reparented wallet stays a +// zombie after it exits. +func processGone(pid int) bool { + if err := syscall.Kill(pid, 0); errors.Is(err, syscall.ESRCH) { + return true + } + out, err := exec.Command("ps", "-o", "stat=", "-p", strconv.Itoa(pid)).Output() + return err == nil && strings.HasPrefix(strings.TrimSpace(string(out)), "Z") +} + +// The daemon dying without stopping its apps (SIGKILL, crash, watchdog +// exit) must not leave the wallet running as an orphan: it has to notice, +// shut down through its normal path and remove its socket. +func TestWalletExitsWhenParentDies(t *testing.T) { + if testing.Short() { + t.Skip("spawns processes") + } + exe, err := os.Executable() + if err != nil { + t.Fatal(err) + } + dir := shortSockDir(t) + sock := filepath.Join(dir, "app.sock") + logPath := filepath.Join(t.TempDir(), "wallet.log") + args, _ := json.Marshal([]string{ + "--addr", "0:0001.0001.0001", + "--db", filepath.Join(dir, "data.db"), + "--socket", sock, + "--identity", filepath.Join(dir, "identity.json"), + "--cap-state", filepath.Join(dir, "cap-state.jsonl"), + "--no-evm", + }) + + parent := exec.Command(exe, "-test.run=^$") + parent.Env = append(os.Environ(), + "WALLET_TEST_ROLE=parent", + "WALLET_TEST_ARGS="+string(args), + "WALLET_TEST_LOG="+logPath, + ) + parent.Stderr = os.Stderr + out, err := parent.StdoutPipe() + if err != nil { + t.Fatal(err) + } + if err := parent.Start(); err != nil { + t.Fatal(err) + } + t.Cleanup(func() { _ = parent.Process.Kill(); _ = parent.Wait() }) + + line, err := bufio.NewReader(out).ReadString('\n') + if err != nil { + t.Fatalf("read wallet pid from fake daemon: %v", err) + } + walletPID, err := strconv.Atoi(strings.TrimSpace(line)) + if err != nil { + t.Fatalf("wallet pid %q: %v", line, err) + } + t.Cleanup(func() { + if !processGone(walletPID) { + _ = syscall.Kill(walletPID, syscall.SIGKILL) + } + }) + + walletLog := func() string { b, _ := os.ReadFile(logPath); return string(b) } + deadline := time.Now().Add(15 * time.Second) + for { + if c, err := net.DialTimeout("unix", sock, 100*time.Millisecond); err == nil { + c.Close() + break + } + if time.Now().After(deadline) { + t.Fatalf("wallet socket never came up; log:\n%s", walletLog()) + } + time.Sleep(20 * time.Millisecond) + } + + // The daemon dies hard. + if err := parent.Process.Kill(); err != nil { + t.Fatal(err) + } + _ = parent.Wait() + killed := time.Now() + + for !processGone(walletPID) { + if time.Since(killed) > 10*time.Second { + t.Fatalf("wallet pid %d still running %s after its parent was killed (orphaned); log:\n%s", + walletPID, time.Since(killed).Round(time.Millisecond), walletLog()) + } + time.Sleep(20 * time.Millisecond) + } + t.Logf("wallet exited %s after its parent was killed", time.Since(killed).Round(time.Millisecond)) + + logs := walletLog() + for _, want := range []string{"exited (reparented to", "graceful shutdown complete"} { + if !strings.Contains(logs, want) { + t.Errorf("wallet log lacks %q; log:\n%s", want, logs) + } + } + if _, err := os.Stat(sock); !os.IsNotExist(err) { + t.Errorf("socket %s left behind after the wallet exited (stat err=%v)", sock, err) + } +} + +// serve must not write a log line per IPC call by default: the supervisor +// wires the wallet's stderr into the daemon log, so per-call lines grow +// that log without bound. +func TestServeLogsConnectionsOnlyWhenAsked(t *testing.T) { + for _, tc := range []struct { + logConns bool + want int // "conn open"/"conn closed" lines for n calls + }{ + {logConns: false, want: 0}, + {logConns: true, want: 2 * 25}, + } { + t.Run(fmt.Sprintf("log-conns=%v", tc.logConns), func(t *testing.T) { + const n = 25 + sock := filepath.Join(shortSockDir(t), "s.sock") + ln, err := net.Listen("unix", sock) + if err != nil { + t.Fatal(err) + } + d := ipc.NewDispatcher() + d.Register("t.ping", func(context.Context, *ipc.Envelope) (json.RawMessage, error) { + return json.RawMessage(`{"ok":true}`), nil + }) + var buf syncBuffer + logger := log.New(&buf, "", 0) + ctx, cancel := context.WithCancel(context.Background()) + done := make(chan error, 1) + go func() { done <- serve(ctx, ln, d, logger, tc.logConns) }() + + for i := 0; i < n; i++ { + c, err := net.Dial("unix", sock) + if err != nil { + t.Fatal(err) + } + var out map[string]any + if err := ipc.Call(c, "t.ping", nil, &out); err != nil { + t.Fatal(err) + } + c.Close() + } + // Let the last connection's goroutine finish logging. + time.Sleep(50 * time.Millisecond) + cancel() + if err := <-done; err != nil { + t.Fatal(err) + } + got := strings.Count(buf.String(), "conn open") + strings.Count(buf.String(), "conn closed") + if got != tc.want { + t.Errorf("%d per-connection log lines for %d calls, want %d; log:\n%s", got, n, tc.want, buf.String()) + } + }) + } +} + +type syncBuffer struct { + mu sync.Mutex + buf bytes.Buffer +} + +func (b *syncBuffer) Write(p []byte) (int, error) { + b.mu.Lock() + defer b.mu.Unlock() + return b.buf.Write(p) +} + +func (b *syncBuffer) String() string { + b.mu.Lock() + defer b.mu.Unlock() + return b.buf.String() +} + +var _ io.Writer = (*syncBuffer)(nil) diff --git a/cmd/wallet/main.go b/cmd/wallet/main.go index 4ef77fe..b215e0b 100644 --- a/cmd/wallet/main.go +++ b/cmd/wallet/main.go @@ -68,19 +68,21 @@ func main() { func run(ctx context.Context, args []string) error { fs := flag.NewFlagSet("wallet", flag.ContinueOnError) var ( - addr = fs.String("addr", "", "pilot address (e.g. 0:0001.HHHH.LLLL); required") - dbPath = fs.String("db", defaultPath("data.db"), "sqlite ledger path") - sockPath = fs.String("socket", defaultPath("wallet.sock"), "unix socket to listen on") - idPath = fs.String("identity", defaultPath("identity.json"), "ed25519 identity file (created on first start)") - mfPath = fs.String("manifest", "", "path to manifest.json; when set, key.sign:cap grants activate runtime spend caps") - capState = fs.String("cap-state", "", "JSONL spend log; persists rolling-window cap state across wallet restarts so caps aren't bypassable by daemon-restart") - evmIDPath = fs.String("evm-identity", defaultPath("identity-evm.json"), "secp256k1 identity for EVM/USDC payments (created on first start)") - evmChains = fs.String("evm-chains", "8453,1,137", "comma-separated EVM chain IDs to enable. First is primary (used when wallet.evm.* requests omit chain_id). Known: 1=Ethereum, 8453=Base, 137=Polygon, 84532=Base Sepolia. $PILOT_EVM_CHAINS overrides this when --evm-chains was left at the default.") - evmRPC = fs.String("evm-rpc", "", "PRIMARY chain's JSON-RPC endpoint. For per-chain endpoints use PILOT_EVM_RPC_ env vars (e.g. PILOT_EVM_RPC_137=https://polygon-rpc.com). Falls back to $PILOT_EVM_RPC for the primary chain.") - evmOff = fs.Bool("no-evm", false, "disable every wallet.evm.* method (no secp256k1 key created)") + addr = fs.String("addr", "", "pilot address (e.g. 0:0001.HHHH.LLLL); required") + dbPath = fs.String("db", defaultPath("data.db"), "sqlite ledger path") + sockPath = fs.String("socket", defaultPath("wallet.sock"), "unix socket to listen on") + idPath = fs.String("identity", defaultPath("identity.json"), "ed25519 identity file (created on first start)") + mfPath = fs.String("manifest", "", "path to manifest.json; when set, key.sign:cap grants activate runtime spend caps") + capState = fs.String("cap-state", "", "JSONL spend log; persists rolling-window cap state across wallet restarts so caps aren't bypassable by daemon-restart") + evmIDPath = fs.String("evm-identity", defaultPath("identity-evm.json"), "secp256k1 identity for EVM/USDC payments (created on first start)") + evmChains = fs.String("evm-chains", "8453,1,137", "comma-separated EVM chain IDs to enable. First is primary (used when wallet.evm.* requests omit chain_id). Known: 1=Ethereum, 8453=Base, 137=Polygon, 84532=Base Sepolia. $PILOT_EVM_CHAINS overrides this when --evm-chains was left at the default.") + evmRPC = fs.String("evm-rpc", "", "PRIMARY chain's JSON-RPC endpoint. For per-chain endpoints use PILOT_EVM_RPC_ env vars (e.g. PILOT_EVM_RPC_137=https://polygon-rpc.com). Falls back to $PILOT_EVM_RPC for the primary chain.") + evmOff = fs.Bool("no-evm", false, "disable every wallet.evm.* method (no secp256k1 key created)") settlerAddr = fs.String("settler-addr", "", "TCP endpoint of the pilot-protocol/settler service (host:port). Empty disables wallet.settler.* methods. Env: PILOT_SETTLER_ADDR.") settlerPubkeyHex = fs.String("settler-pubkey", "", "expected settler ed25519 pubkey (hex) — when set, the wallet refuses to start if the live settler advertises a different pubkey. Env: PILOT_SETTLER_PUBKEY.") - showVer = fs.Bool("version", false, "print version and exit") + showVer = fs.Bool("version", false, "print version and exit") + exitWithParent = fs.Bool("exit-with-parent", true, "shut down when the process that started the wallet (the pilot daemon) exits, instead of lingering as an orphan") + logConns = fs.Bool("log-conns", false, "log every IPC connection open/close (one pair of lines per call; off by default so the daemon log does not grow with call volume)") ) fs.SetOutput(os.Stderr) if err := fs.Parse(args); err != nil { @@ -99,6 +101,12 @@ func run(ctx context.Context, args []string) error { } logger := log.New(os.Stderr, "wallet ", log.LstdFlags|log.Lmicroseconds) + + ctx, cancel := context.WithCancel(ctx) + defer cancel() + if *exitWithParent && watchParent(ctx, cancel, startPPID, parentPollInterval, logger) { + logger.Printf("watching parent pid %d", startPPID) + } logger.Printf("starting addr=%s db=%s socket=%s identity=%s", *addr, *dbPath, *sockPath, *idPath) signer, err := wallet.LoadOrCreateLocalSigner(*idPath) @@ -265,11 +273,14 @@ func run(ctx context.Context, args []string) error { // is created as 0o600 atomically — no TOCTOU window between // socket creation and a post-hoc Chmod. oldMask := syscall.Umask(0o177) - listener, err := net.Listen("unix", *sockPath) + ln, err := net.Listen("unix", *sockPath) syscall.Umask(oldMask) if err != nil { return fmt.Errorf("listen %s: %w", *sockPath, err) } + // Remember which socket file we bound, so shutdown never unlinks a + // socket a replacement instance has bound at the same path since. + listener := ownSocket(ln.(*net.UnixListener), *sockPath, logger) // Chmod is a belt-and-suspenders backup; the umask above covers // the primary case. On the off-chance a platform doesn't apply // umask to unix sockets, the explicit chmod is the fallback. @@ -278,18 +289,18 @@ func run(ctx context.Context, args []string) error { } logger.Printf("listening on %s methods=%d", *sockPath, len(dispatcher.Methods())) - return serve(ctx, listener, dispatcher, logger) + return serve(ctx, listener, dispatcher, logger, *logConns) } // serve is the accept loop. Returns nil on clean shutdown. -func serve(ctx context.Context, listener net.Listener, d *ipc.Dispatcher, logger *log.Logger) error { +func serve(ctx context.Context, listener net.Listener, d *ipc.Dispatcher, logger *log.Logger, logConns bool) error { // Close the listener when ctx is canceled. On macOS Close() alone // doesn't always unblock a pending Accept(); a past-deadline pokes it. go func() { <-ctx.Done() logger.Printf("shutting down") - if ul, ok := listener.(*net.UnixListener); ok { - _ = ul.SetDeadline(time.Now()) + if dl, ok := listener.(interface{ SetDeadline(time.Time) error }); ok { + _ = dl.SetDeadline(time.Now()) } _ = listener.Close() }() @@ -326,11 +337,15 @@ acceptLoop: case <-stop: } }() - logger.Printf("conn open from=%s", c.RemoteAddr()) + if logConns { + logger.Printf("conn open from=%s", c.RemoteAddr()) + } if err := ipc.Serve(ctx, c, d); err != nil { logger.Printf("serve: %v", err) } - logger.Printf("conn closed") + if logConns { + logger.Printf("conn closed") + } }(conn) } diff --git a/cmd/wallet/parentwatch.go b/cmd/wallet/parentwatch.go new file mode 100644 index 0000000..135e285 --- /dev/null +++ b/cmd/wallet/parentwatch.go @@ -0,0 +1,61 @@ +package main + +import ( + "context" + "log" + "os" + "time" +) + +// parentPollInterval is how often the wallet checks whether the process +// that spawned it is still alive. One getppid(2) per second is free, and +// bounds how long an orphaned wallet keeps its socket, ledger and +// cap-state file open after the daemon is gone. +const parentPollInterval = time.Second + +// getppid is swapped out by tests. +var getppid = os.Getppid + +// startPPID is the parent pid captured as early as possible in main(), so +// a daemon that dies while the wallet is still starting up (opening the +// ledger, loading keys) is still noticed. +var startPPID = os.Getppid() + +// watchParent cancels the wallet's run context once the process that +// spawned it exits, so the wallet shuts down through its normal SIGTERM +// path (stop accepting, drain, close the ledger, unlink the socket). +// +// The pilot daemon's app-store supervisor starts each app in its own +// process group. On Linux it also sets Pdeathsig=SIGKILL, but macOS has +// no parent-death signal, and anything that bypasses the daemon's +// shutdown path (the rx watchdog's os.Exit, SIGKILL, a crash, launchd +// restarting the service) leaves the wallet reparented to launchd, still +// serving its socket and holding its sqlite ledger and cap-state log open +// next to the instance the respawned daemon starts. Watching our own +// parent closes that gap on every platform without cooperation from the +// supervisor. +// +// A wallet started directly by init/launchd (ppid 1 at startup) has no +// parent to watch and is left alone. +func watchParent(ctx context.Context, cancel context.CancelFunc, ppid int, interval time.Duration, logger *log.Logger) bool { + if ppid <= 1 { + return false + } + go func() { + t := time.NewTicker(interval) + defer t.Stop() + for { + if now := getppid(); now != ppid { + logger.Printf("parent pid %d exited (reparented to %d); shutting down", ppid, now) + cancel() + return + } + select { + case <-ctx.Done(): + return + case <-t.C: + } + } + }() + return true +} diff --git a/cmd/wallet/parentwatch_test.go b/cmd/wallet/parentwatch_test.go new file mode 100644 index 0000000..c14b92d --- /dev/null +++ b/cmd/wallet/parentwatch_test.go @@ -0,0 +1,50 @@ +package main + +import ( + "context" + "io" + "log" + "sync/atomic" + "testing" + "time" +) + +func TestWatchParentCancelsWhenReparented(t *testing.T) { + var ppid atomic.Int64 + ppid.Store(4242) + orig := getppid + getppid = func() int { return int(ppid.Load()) } + t.Cleanup(func() { getppid = orig }) + + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + if !watchParent(ctx, cancel, 4242, 5*time.Millisecond, log.New(io.Discard, "", 0)) { + t.Fatal("watchParent refused a real parent pid") + } + + select { + case <-ctx.Done(): + t.Fatal("canceled while the parent was still alive") + case <-time.After(50 * time.Millisecond): + } + + ppid.Store(1) // parent died, reparented to init/launchd + select { + case <-ctx.Done(): + case <-time.After(2 * time.Second): + t.Fatal("ctx not canceled after the parent exited") + } +} + +func TestWatchParentSkipsInitParent(t *testing.T) { + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + for _, p := range []int{0, 1} { + if watchParent(ctx, cancel, p, time.Millisecond, log.New(io.Discard, "", 0)) { + t.Fatalf("watchParent(ppid=%d) armed; a wallet started by init has no parent to follow", p) + } + } + if ctx.Err() != nil { + t.Fatal("ctx canceled") + } +} diff --git a/cmd/wallet/socketguard.go b/cmd/wallet/socketguard.go new file mode 100644 index 0000000..ccccf4f --- /dev/null +++ b/cmd/wallet/socketguard.go @@ -0,0 +1,57 @@ +package main + +import ( + "log" + "net" + "os" +) + +// ownedUnixListener is the wallet's listener. Closing it unlinks the +// socket path only while that path still names the socket this process +// bound. +// +// Go's UnixListener.Close unlinks the path by name. That is wrong once +// another wallet instance has taken the path over, which is exactly what +// happens when an instance shuts down late: the daemon (or the app-store +// supervisor after a daemon restart) starts a replacement, which removes +// the old socket file and binds a new one at the same path, and then the +// old instance exits — after noticing its parent died, or on a SIGTERM +// that arrives after the replacement is up. Without this check the old +// instance's Close deletes the replacement's socket and leaves the live +// wallet running but unreachable. +type ownedUnixListener struct { + *net.UnixListener + path string + bound os.FileInfo // the socket file as bound; nil if it could not be stat'ed + logger *log.Logger +} + +// ownSocket wraps l so that Close leaves path alone once it no longer +// refers to l's socket. Call it right after Listen. +func ownSocket(l *net.UnixListener, path string, logger *log.Logger) *ownedUnixListener { + fi, err := os.Stat(path) + if err != nil { + fi = nil // cannot tell later; fall back to Go's default unlink-on-close + } + return &ownedUnixListener{UnixListener: l, path: path, bound: fi, logger: logger} +} + +// stillOurs reports whether path still names the socket this listener +// bound. +func (o *ownedUnixListener) stillOurs() bool { + if o.bound == nil { + return true + } + fi, err := os.Stat(o.path) + return err == nil && os.SameFile(o.bound, fi) +} + +func (o *ownedUnixListener) Close() error { + if !o.stillOurs() { + o.UnixListener.SetUnlinkOnClose(false) + if o.logger != nil { + o.logger.Printf("socket %s now belongs to another instance; leaving it in place", o.path) + } + } + return o.UnixListener.Close() +} diff --git a/cmd/wallet/socketguard_test.go b/cmd/wallet/socketguard_test.go new file mode 100644 index 0000000..27b4ff7 --- /dev/null +++ b/cmd/wallet/socketguard_test.go @@ -0,0 +1,102 @@ +package main + +import ( + "io" + "log" + "net" + "os" + "path/filepath" + "testing" + "time" +) + +// shortSockDir returns a directory short enough for a unix socket path +// (macOS sun_path is 104 bytes; t.TempDir under /var/folders can exceed it). +func shortSockDir(t *testing.T) string { + t.Helper() + d, err := os.MkdirTemp("", "wsk") + if err != nil { + t.Fatal(err) + } + t.Cleanup(func() { os.RemoveAll(d) }) + return d +} + +func listenOwned(t *testing.T, path string) *ownedUnixListener { + t.Helper() + l, err := net.Listen("unix", path) + if err != nil { + t.Fatalf("listen %s: %v", path, err) + } + return ownSocket(l.(*net.UnixListener), path, log.New(io.Discard, "", 0)) +} + +// An instance that shuts down after a replacement has taken over its +// socket path must not delete the replacement's socket. +func TestOwnedListenerCloseKeepsReplacementSocket(t *testing.T) { + path := filepath.Join(shortSockDir(t), "app.sock") + old := listenOwned(t, path) + + // The replacement does what the supervisor and the wallet's own + // startup do: drop the stale socket file, then bind a fresh one. + if err := os.Remove(path); err != nil { + t.Fatal(err) + } + repl, err := net.Listen("unix", path) + if err != nil { + t.Fatalf("replacement listen: %v", err) + } + defer repl.Close() + + if err := old.Close(); err != nil { + t.Fatalf("old close: %v", err) + } + + if _, err := os.Stat(path); err != nil { + t.Fatalf("replacement's socket was removed by the old instance's Close: %v", err) + } + accepted := make(chan error, 1) + go func() { + c, err := repl.Accept() + if err == nil { + c.Close() + } + accepted <- err + }() + c, err := net.DialTimeout("unix", path, time.Second) + if err != nil { + t.Fatalf("dial replacement: %v", err) + } + c.Close() + if err := <-accepted; err != nil { + t.Fatalf("replacement accept: %v", err) + } +} + +// Normal shutdown still removes the wallet's own socket. +func TestOwnedListenerCloseUnlinksOwnSocket(t *testing.T) { + path := filepath.Join(shortSockDir(t), "app.sock") + l := listenOwned(t, path) + if err := l.Close(); err != nil { + t.Fatal(err) + } + if _, err := os.Stat(path); !os.IsNotExist(err) { + t.Fatalf("socket still present after close (stat err=%v)", err) + } +} + +// If the socket file is already gone (someone removed it and nobody +// re-bound), Close must not fail or recreate anything. +func TestOwnedListenerCloseAfterExternalRemove(t *testing.T) { + path := filepath.Join(shortSockDir(t), "app.sock") + l := listenOwned(t, path) + if err := os.Remove(path); err != nil { + t.Fatal(err) + } + if err := l.Close(); err != nil { + t.Fatalf("close: %v", err) + } + if _, err := os.Stat(path); !os.IsNotExist(err) { + t.Fatalf("path reappeared (stat err=%v)", err) + } +}