From 98f0e496f61e17441db503e6a2544cb534c3b812 Mon Sep 17 00:00:00 2001 From: Torkel Rogstad Date: Wed, 9 Sep 2026 11:25:06 +0200 Subject: [PATCH 1/2] Implement SSH tunnel self repair --- main.go | 139 ++++++++++++++++++++++++++++++++++++++++---------------- 1 file changed, 99 insertions(+), 40 deletions(-) diff --git a/main.go b/main.go index 95d185e..149c4d4 100644 --- a/main.go +++ b/main.go @@ -4,6 +4,7 @@ import ( "bufio" "context" "fmt" + "io" "net" "os" "os/exec" @@ -37,7 +38,7 @@ func realMain(cfg *config) error { Msgf("setting up SSH tunnel: %d:localhost:%d -> %s", cfg.SSH.LocalPort, cfg.SSH.RemotePort, cfg.SSH.Host, ) - if err := setupSSHTunnel(ctx, cfg.SSH, errs); err != nil { + if err := setupSSHTunnel(ctx, cfg.SSH); err != nil { return fmt.Errorf("setup SSH tunnel: %w", err) } } @@ -110,8 +111,10 @@ func findSetting(key string, settings []debug.BuildSetting) string { return "unknown" } -// setupSSHTunnel creates an SSH tunnel by running the ssh command -func setupSSHTunnel(ctx context.Context, conf sshConfig, out chan error) error { +// setupSSHTunnel starts an ssh port forward and keeps it running until ctx is +// done. Only the initial connection can fail the call. Later exits are +// handled by superviseSSHTunnel. +func setupSSHTunnel(ctx context.Context, conf sshConfig) error { if conf.KeyFile == "" { return fmt.Errorf("ssh: key file is required") } @@ -125,6 +128,8 @@ func setupSSHTunnel(ctx context.Context, conf sshConfig, out chan error) error { "-o", "ServerAliveInterval=60", // send keep-alive every 60 seconds "-o", "ServerAliveCountMax=3", // allow 3 missed keep-alive responses before disconnecting "-o", "TCPKeepAlive=yes", // enable TCP keep-alive + "-o", "ConnectTimeout=10", // never hang in connect, so the supervisor can retry + "-o", "ExitOnForwardFailure=yes", // a tunnel that can't bind the local port is useless, exit and retry "-i", conf.KeyFile, // specify the key file to use "-L", fmt.Sprintf("%d:localhost:%d", conf.LocalPort, conf.RemotePort), conf.Host, @@ -147,58 +152,65 @@ func setupSSHTunnel(ctx context.Context, conf sshConfig, out chan error) error { args = append(args, "-o", "UserKnownHostsFile="+tempFile.Name()) } - // Build SSH command with port forwarding - // -N: Don't execute remote command (forward only) - // -L: Local port forwarding + tunnel, err := startSSHTunnel(ctx, args) + if err != nil { + return err + } + if err := waitForSSHTunnel(ctx, conf.LocalPort, tunnel); err != nil { + return err + } + + go superviseSSHTunnel(ctx, conf.LocalPort, args, tunnel) + + return nil +} + +type sshTunnel struct { + done chan struct{} // closed once ssh has exited + err error // exit error, set before done is closed +} + +func startSSHTunnel(ctx context.Context, args []string) (*sshTunnel, error) { cmd := exec.CommandContext(ctx, "ssh", args...) - // Capture stdout and stderr stdout, err := cmd.StdoutPipe() if err != nil { - return fmt.Errorf("create stdout pipe: %w", err) + return nil, fmt.Errorf("create stdout pipe: %w", err) } stderr, err := cmd.StderrPipe() if err != nil { - return fmt.Errorf("create stderr pipe: %w", err) + return nil, fmt.Errorf("create stderr pipe: %w", err) } - // Start the SSH tunnel if err := cmd.Start(); err != nil { - return fmt.Errorf("starting SSH tunnel: %w", err) + return nil, fmt.Errorf("starting SSH tunnel: %w", err) } - // Monitor the tunnel process in background - go func() { - if err := cmd.Wait(); err != nil { - zerolog.Ctx(ctx).Error(). - Err(err). - Msg("SSH tunnel exited unexpectedly") - out <- fmt.Errorf("SSH tunnel exited unexpectedly: %w", err) - } - }() + go logSSHOutput(ctx, "stdout", stdout) + go logSSHOutput(ctx, "stderr", stderr) - // Log stdout in background + tunnel := &sshTunnel{done: make(chan struct{})} go func() { - scanner := bufio.NewScanner(stdout) - for scanner.Scan() { - zerolog.Ctx(ctx).Debug(). - Msgf("SSH tunnel stdout: %s", scanner.Text()) - } + tunnel.err = cmd.Wait() + close(tunnel.done) }() - // Log stderr in background - go func() { - scanner := bufio.NewScanner(stderr) - for scanner.Scan() { - zerolog.Ctx(ctx).Debug(). - Msgf("SSH tunnel stderr: %s", scanner.Text()) - } - }() + return tunnel, nil +} - // Wait for the tunnel to be established - for i := 0; i < 10; i++ { - if conn, err := net.Dial("tcp", fmt.Sprintf("localhost:%d", conf.LocalPort)); err == nil { +func logSSHOutput(ctx context.Context, name string, r io.Reader) { + scanner := bufio.NewScanner(r) + for scanner.Scan() { + zerolog.Ctx(ctx).Debug(). + Msgf("SSH tunnel %s: %s", name, scanner.Text()) + } +} + +// waitForSSHTunnel blocks until the local port accepts connections. +func waitForSSHTunnel(ctx context.Context, localPort int, tunnel *sshTunnel) error { + for range 10 { + if conn, err := net.Dial("tcp", fmt.Sprintf("localhost:%d", localPort)); err == nil { if err := conn.Close(); err != nil { return fmt.Errorf("close connection: %w", err) } @@ -207,12 +219,59 @@ func setupSSHTunnel(ctx context.Context, conf sshConfig, out chan error) error { select { case <-ctx.Done(): return fmt.Errorf("wait for SSH tunnel: %w", ctx.Err()) - case err := <-out: - return fmt.Errorf("setup SSH tunnel: %w", err) - + case <-tunnel.done: + return fmt.Errorf("SSH tunnel exited: %w", tunnel.err) case <-time.After(time.Second): } } return fmt.Errorf("timeout waiting for SSH tunnel") } + +// superviseSSHTunnel restarts ssh with backoff whenever it exits, until ctx +// is done. +func superviseSSHTunnel(ctx context.Context, localPort int, args []string, tunnel *sshTunnel) { + log := zerolog.Ctx(ctx) + + const minBackoff, maxBackoff = time.Second, 30 * time.Second + backoff := minBackoff + started := time.Now() + for { + select { + case <-ctx.Done(): + return + case <-tunnel.done: + } + if ctx.Err() != nil { + return + } + + // A tunnel that held for a while earns a fresh backoff. + if time.Since(started) > time.Minute { + backoff = minBackoff + } + log.Error().Err(tunnel.err). + Dur("backoff", backoff). + Msg("SSH tunnel exited, restarting") + + select { + case <-ctx.Done(): + return + case <-time.After(backoff): + } + backoff = min(backoff*2, maxBackoff) + + next, err := startSSHTunnel(ctx, args) + if err != nil { + log.Err(err).Msg("restart SSH tunnel") + continue + } + tunnel, started = next, time.Now() + + if err := waitForSSHTunnel(ctx, localPort, tunnel); err != nil { + // Either ssh already exited (done is closed, the loop restarts it) + // or it is still connecting, which ConnectTimeout bounds. + log.Err(err).Msg("restart SSH tunnel") + } + } +} From ea397e42e9619b0ef21a1002d78cf49a44ba7786 Mon Sep 17 00:00:00 2001 From: Torkel Rogstad Date: Sat, 19 Sep 2026 11:34:45 +0200 Subject: [PATCH 2/2] Make RPCs wait while tunnel is being created --- main.go | 160 ++++++++++++++++--- main_test.go | 389 +++++++++++++++++++++++++++++++++++++++++++++++ server/server.go | 28 +++- 3 files changed, 551 insertions(+), 26 deletions(-) create mode 100644 main_test.go diff --git a/main.go b/main.go index 149c4d4..49a67b5 100644 --- a/main.go +++ b/main.go @@ -3,6 +3,7 @@ package main import ( "bufio" "context" + "errors" "fmt" "io" "net" @@ -10,8 +11,10 @@ import ( "os/exec" "os/signal" "runtime/debug" + "sync" "time" + "connectrpc.com/connect" "github.com/rs/zerolog" "github.com/barebitcoin/btc-buf/server" @@ -33,20 +36,22 @@ func realMain(cfg *config) error { }() errs := make(chan error) + var opts []server.Option if cfg.SSH.Host != "" { zerolog.Ctx(ctx).Info(). Msgf("setting up SSH tunnel: %d:localhost:%d -> %s", cfg.SSH.LocalPort, cfg.SSH.RemotePort, cfg.SSH.Host, ) - if err := setupSSHTunnel(ctx, cfg.SSH); err != nil { + gate := newTunnelGate() + if err := setupSSHTunnel(ctx, cfg.SSH, gate); err != nil { return fmt.Errorf("setup SSH tunnel: %w", err) } + opts = append(opts, server.WithInterceptors(gate.interceptor(tunnelRepairWait))) } clientCtx, clientCancel := context.WithTimeout(ctx, time.Second*10) defer clientCancel() - var opts []server.Option if cfg.AllowPrivateDescriptorsExport { zerolog.Ctx(ctx).Info().Msg("allowing private descriptors export") opts = append(opts, server.WithAllowPrivateDescriptorsExport()) @@ -114,7 +119,7 @@ func findSetting(key string, settings []debug.BuildSetting) string { // setupSSHTunnel starts an ssh port forward and keeps it running until ctx is // done. Only the initial connection can fail the call. Later exits are // handled by superviseSSHTunnel. -func setupSSHTunnel(ctx context.Context, conf sshConfig) error { +func setupSSHTunnel(ctx context.Context, conf sshConfig, gate *tunnelGate) error { if conf.KeyFile == "" { return fmt.Errorf("ssh: key file is required") } @@ -156,22 +161,127 @@ func setupSSHTunnel(ctx context.Context, conf sshConfig) error { if err != nil { return err } - if err := waitForSSHTunnel(ctx, conf.LocalPort, tunnel); err != nil { + waitCtx, cancel := context.WithTimeout(ctx, 10*time.Second) + defer cancel() + if err := waitForSSHTunnel(waitCtx, conf.LocalPort, tunnel); err != nil { return err } + gate.set(true) - go superviseSSHTunnel(ctx, conf.LocalPort, args, tunnel) + go superviseSSHTunnel(ctx, conf.LocalPort, args, tunnel, gate, tunnelReadyTimeout) return nil } +// sshBinary is a variable so tests can substitute a script. +var sshBinary = "ssh" + type sshTunnel struct { done chan struct{} // closed once ssh has exited err error // exit error, set before done is closed + kill func() error +} + +const ( + // tunnelRepairWait bounds how long a request waits for the supervisor to + // bring the tunnel back before failing with Unavailable. + tunnelRepairWait = 15 * time.Second + + // tunnelReadyTimeout bounds how long the supervisor lets a restarted ssh + // take to bind the local port before killing it. ConnectTimeout only + // covers the TCP connect and key exchange, not authentication. + tunnelReadyTimeout = 30 * time.Second +) + +// tunnelGate tracks whether the local ssh port forward accepts connections. +// It says nothing about the remote end of the forward. The tunnel code flips +// it; requests only wait on it. +type tunnelGate struct { + mu sync.Mutex + up bool + gen uint64 // incremented each time the tunnel comes up + changed chan struct{} // closed and replaced on every transition +} + +func newTunnelGate() *tunnelGate { + return &tunnelGate{changed: make(chan struct{})} +} + +func (g *tunnelGate) set(up bool) { + g.mu.Lock() + defer g.mu.Unlock() + + if g.up == up { + return + } + g.up = up + if up { + g.gen++ + } + close(g.changed) + g.changed = make(chan struct{}) +} + +// waitUp blocks until the tunnel is up with a generation newer than after, +// and returns that generation. +func (g *tunnelGate) waitUp(ctx context.Context, after uint64) (uint64, error) { + for { + g.mu.Lock() + up, gen, changed := g.up, g.gen, g.changed + g.mu.Unlock() + + if up && gen > after { + return gen, nil + } + select { + case <-changed: + case <-ctx.Done(): + return 0, ctx.Err() + } + } +} + +// interceptor holds requests while the tunnel is down, for at most wait. A +// request refused by a tunnel that died under it never reached Bitcoin Core, +// and is retried once on the next tunnel generation. +func (g *tunnelGate) interceptor(wait time.Duration) connect.Interceptor { + return connect.UnaryInterceptorFunc(func(next connect.UnaryFunc) connect.UnaryFunc { + return func(ctx context.Context, req connect.AnyRequest) (connect.AnyResponse, error) { + waitCtx, cancel := context.WithTimeout(ctx, wait) + defer cancel() + + var seen uint64 + retried := false + for { + gen, err := g.waitUp(waitCtx, seen) + if err != nil { + // The caller gave up; connect maps this to Canceled or + // DeadlineExceeded. + if ctx.Err() != nil { + return nil, err + } + zerolog.Ctx(ctx).Warn(). + Dur("waited", wait). + Msg("SSH tunnel still down, giving up on request") + return nil, connect.NewError(connect.CodeUnavailable, + fmt.Errorf("SSH tunnel to Bitcoin Core is down: %w", err)) + } + + res, err := next(ctx, req) + if !errors.Is(err, server.ErrUnreachable) || retried { + return res, err + } + + zerolog.Ctx(ctx).Warn().Err(err). + Msg("request hit a dead SSH tunnel, waiting for repair") + seen, retried = gen, true + } + } + }) } func startSSHTunnel(ctx context.Context, args []string) (*sshTunnel, error) { - cmd := exec.CommandContext(ctx, "ssh", args...) + cmd := exec.CommandContext(ctx, sshBinary, args...) stdout, err := cmd.StdoutPipe() if err != nil { @@ -190,7 +300,7 @@ func startSSHTunnel(ctx context.Context, args []string) (*sshTunnel, error) { go logSSHOutput(ctx, "stdout", stdout) go logSSHOutput(ctx, "stderr", stderr) - tunnel := &sshTunnel{done: make(chan struct{})} + tunnel := &sshTunnel{done: make(chan struct{}), kill: cmd.Process.Kill} go func() { tunnel.err = cmd.Wait() close(tunnel.done) @@ -207,9 +317,10 @@ func logSSHOutput(ctx context.Context, name string, r io.Reader) { } } -// waitForSSHTunnel blocks until the local port accepts connections. +// waitForSSHTunnel blocks until the local port accepts connections, ssh +// exits, or ctx is done. func waitForSSHTunnel(ctx context.Context, localPort int, tunnel *sshTunnel) error { - for range 10 { + for { if conn, err := net.Dial("tcp", fmt.Sprintf("localhost:%d", localPort)); err == nil { if err := conn.Close(); err != nil { return fmt.Errorf("close connection: %w", err) @@ -224,16 +335,20 @@ func waitForSSHTunnel(ctx context.Context, localPort int, tunnel *sshTunnel) err case <-time.After(time.Second): } } - - return fmt.Errorf("timeout waiting for SSH tunnel") } // superviseSSHTunnel restarts ssh with backoff whenever it exits, until ctx -// is done. -func superviseSSHTunnel(ctx context.Context, localPort int, args []string, tunnel *sshTunnel) { +// is done. A restarted ssh that has not bound the local port within +// readyTimeout is killed and restarted. +func superviseSSHTunnel( + ctx context.Context, localPort int, args []string, + tunnel *sshTunnel, gate *tunnelGate, readyTimeout time.Duration, +) { log := zerolog.Ctx(ctx) - const minBackoff, maxBackoff = time.Second, 30 * time.Second + // maxBackoff stays below tunnelRepairWait so a held request sees at + // least one reconnect attempt. + const minBackoff, maxBackoff = time.Second, 10 * time.Second backoff := minBackoff started := time.Now() for { @@ -245,6 +360,7 @@ func superviseSSHTunnel(ctx context.Context, localPort int, args []string, tunne if ctx.Err() != nil { return } + gate.set(false) // A tunnel that held for a while earns a fresh backoff. if time.Since(started) > time.Minute { @@ -268,10 +384,18 @@ func superviseSSHTunnel(ctx context.Context, localPort int, args []string, tunne } tunnel, started = next, time.Now() - if err := waitForSSHTunnel(ctx, localPort, tunnel); err != nil { - // Either ssh already exited (done is closed, the loop restarts it) - // or it is still connecting, which ConnectTimeout bounds. - log.Err(err).Msg("restart SSH tunnel") + readyCtx, cancel := context.WithTimeout(ctx, readyTimeout) + err = waitForSSHTunnel(readyCtx, localPort, tunnel) + cancel() + if err != nil { + if errors.Is(err, context.DeadlineExceeded) { + log.Err(err).Msg("SSH tunnel not ready in time, killing it") + if err := tunnel.kill(); err != nil { + log.Err(err).Msg("kill SSH tunnel") + } + } + continue } + gate.set(true) } } diff --git a/main_test.go b/main_test.go new file mode 100644 index 0000000..3523ecd --- /dev/null +++ b/main_test.go @@ -0,0 +1,389 @@ +package main + +import ( + "context" + "errors" + "fmt" + "net" + "os" + "path/filepath" + "strings" + "sync" + "testing" + "time" + + "connectrpc.com/connect" + "github.com/rs/zerolog" + + "github.com/barebitcoin/btc-buf/server" +) + +type fakeNext struct { + mu sync.Mutex + calls int + fn func(call int) error +} + +func (f *fakeNext) unary(context.Context, connect.AnyRequest) (connect.AnyResponse, error) { + f.mu.Lock() + f.calls++ + call := f.calls + f.mu.Unlock() + + if err := f.fn(call); err != nil { + return nil, err + } + return connect.NewResponse(&struct{}{}), nil +} + +func (f *fakeNext) count() int { + f.mu.Lock() + defer f.mu.Unlock() + return f.calls +} + +func callThroughGate(ctx context.Context, gate *tunnelGate, wait time.Duration, next *fakeNext) error { + handler := gate.interceptor(wait).WrapUnary(next.unary) + _, err := handler(ctx, connect.NewRequest(&struct{}{})) + return err +} + +func refused() error { + return connect.NewError(connect.CodeUnavailable, server.ErrUnreachable) +} + +func isUp(gate *tunnelGate) bool { + ctx, cancel := context.WithTimeout(context.Background(), 5*time.Millisecond) + defer cancel() + _, err := gate.waitUp(ctx, 0) + return err == nil +} + +func TestTunnelGate_PassesThroughWhenUp(t *testing.T) { + gate := newTunnelGate() + gate.set(true) + + next := &fakeNext{fn: func(int) error { return nil }} + if err := callThroughGate(context.Background(), gate, time.Second, next); err != nil { + t.Fatal(err) + } + if next.count() != 1 { + t.Fatalf("expected 1 call, got %d", next.count()) + } +} + +func TestTunnelGate_HoldsRequestUntilUp(t *testing.T) { + gate := newTunnelGate() + next := &fakeNext{fn: func(int) error { return nil }} + + done := make(chan error, 1) + go func() { done <- callThroughGate(context.Background(), gate, 5*time.Second, next) }() + + select { + case err := <-done: + t.Fatalf("request completed while tunnel was down: %v", err) + case <-time.After(50 * time.Millisecond): + } + + gate.set(true) + if err := <-done; err != nil { + t.Fatal(err) + } + if next.count() != 1 { + t.Fatalf("expected 1 call, got %d", next.count()) + } +} + +func TestTunnelGate_ReleasesAllWaiters(t *testing.T) { + gate := newTunnelGate() + next := &fakeNext{fn: func(int) error { return nil }} + + const waiters = 20 + errs := make(chan error, waiters) + for range waiters { + go func() { errs <- callThroughGate(context.Background(), gate, 5*time.Second, next) }() + } + + time.Sleep(30 * time.Millisecond) + gate.set(true) + + for range waiters { + if err := <-errs; err != nil { + t.Fatal(err) + } + } + if next.count() != waiters { + t.Fatalf("expected %d calls, got %d", waiters, next.count()) + } +} + +func TestTunnelGate_FailsWhenTunnelStaysDown(t *testing.T) { + gate := newTunnelGate() + next := &fakeNext{fn: func(int) error { return nil }} + + err := callThroughGate(context.Background(), gate, 50*time.Millisecond, next) + if connect.CodeOf(err) != connect.CodeUnavailable { + t.Fatalf("expected unavailable, got %v", err) + } + if !errors.Is(err, context.DeadlineExceeded) { + t.Fatalf("expected deadline exceeded cause, got %v", err) + } + if next.count() != 0 { + t.Fatalf("expected no calls, got %d", next.count()) + } +} + +func TestTunnelGate_CallerCancelWhileHeld(t *testing.T) { + gate := newTunnelGate() + next := &fakeNext{fn: func(int) error { return nil }} + + ctx, cancel := context.WithCancel(context.Background()) + go func() { + time.Sleep(20 * time.Millisecond) + cancel() + }() + + err := callThroughGate(ctx, gate, 5*time.Second, next) + if !errors.Is(err, context.Canceled) { + t.Fatalf("expected context canceled, got %v", err) + } + // Left uncoded on purpose: connect maps it to CodeCanceled. + if connect.CodeOf(err) == connect.CodeUnavailable { + t.Fatalf("caller cancellation must not be reported as unavailable: %v", err) + } + if next.count() != 0 { + t.Fatalf("expected no calls, got %d", next.count()) + } +} + +func TestTunnelGate_SlowRPCNotBoundedByWait(t *testing.T) { + gate := newTunnelGate() + gate.set(true) + + next := &fakeNext{fn: func(int) error { + time.Sleep(60 * time.Millisecond) + return nil + }} + if err := callThroughGate(context.Background(), gate, 20*time.Millisecond, next); err != nil { + t.Fatal(err) + } +} + +func TestTunnelGate_RetriesRefusedAfterRepair(t *testing.T) { + gate := newTunnelGate() + gate.set(true) + + next := &fakeNext{} + next.fn = func(call int) error { + if call > 1 { + return nil + } + return refused() + } + + // The supervisor notices the dead ssh a bit later and brings up a new one. + go func() { + time.Sleep(30 * time.Millisecond) + gate.set(false) + time.Sleep(30 * time.Millisecond) + gate.set(true) + }() + + if err := callThroughGate(context.Background(), gate, time.Second, next); err != nil { + t.Fatal(err) + } + if next.count() != 2 { + t.Fatalf("expected 2 calls, got %d", next.count()) + } +} + +func TestTunnelGate_RetriesRefusedOnlyOnce(t *testing.T) { + gate := newTunnelGate() + gate.set(true) + + next := &fakeNext{fn: func(int) error { + gate.set(false) + gate.set(true) + return refused() + }} + + err := callThroughGate(context.Background(), gate, time.Second, next) + if !errors.Is(err, server.ErrUnreachable) { + t.Fatalf("expected refused error, got %v", err) + } + if next.count() != 2 { + t.Fatalf("expected 2 calls, got %d", next.count()) + } +} + +func TestTunnelGate_RefusedWithoutRepairFails(t *testing.T) { + gate := newTunnelGate() + gate.set(true) + + next := &fakeNext{fn: func(int) error { return refused() }} + + err := callThroughGate(context.Background(), gate, 50*time.Millisecond, next) + if connect.CodeOf(err) != connect.CodeUnavailable { + t.Fatalf("expected unavailable, got %v", err) + } + if !errors.Is(err, context.DeadlineExceeded) { + t.Fatalf("expected deadline exceeded cause, got %v", err) + } + if next.count() != 1 { + t.Fatalf("expected 1 call, got %d", next.count()) + } +} + +func TestTunnelGate_OtherErrorsPassThrough(t *testing.T) { + gate := newTunnelGate() + gate.set(true) + + boom := errors.New("boom") + next := &fakeNext{fn: func(int) error { return boom }} + + if err := callThroughGate(context.Background(), gate, time.Second, next); !errors.Is(err, boom) { + t.Fatalf("expected boom, got %v", err) + } + if next.count() != 1 { + t.Fatalf("expected 1 call, got %d", next.count()) + } +} + +func TestTunnelGate_SetIsIdempotent(t *testing.T) { + gate := newTunnelGate() + gate.set(true) + gate.set(true) + + gen, err := gate.waitUp(context.Background(), 0) + if err != nil { + t.Fatal(err) + } + if gen != 1 { + t.Fatalf("expected generation 1, got %d", gen) + } + + gate.set(false) + gate.set(false) + gate.set(true) + + gen, err = gate.waitUp(context.Background(), 1) + if err != nil { + t.Fatal(err) + } + if gen != 2 { + t.Fatalf("expected generation 2, got %d", gen) + } +} + +func freePort(t *testing.T) int { + t.Helper() + l, err := net.Listen("tcp", "127.0.0.1:0") + if err != nil { + t.Fatal(err) + } + port := l.Addr().(*net.TCPAddr).Port + if err := l.Close(); err != nil { + t.Fatal(err) + } + return port +} + +func spawnCount(t *testing.T, path string) int { + t.Helper() + data, err := os.ReadFile(path) + if errors.Is(err, os.ErrNotExist) { + return 0 + } + if err != nil { + t.Fatal(err) + } + return strings.Count(string(data), "spawn") +} + +func waitFor(t *testing.T, timeout time.Duration, what string, cond func() bool) { + t.Helper() + start := time.Now() + for !cond() { + if time.Since(start) > timeout { + t.Fatalf("timed out waiting for %s", what) + } + time.Sleep(10 * time.Millisecond) + } + t.Logf("%s after %s", what, time.Since(start).Round(time.Millisecond)) +} + +// Drives superviseSSHTunnel with a shell script standing in for ssh: the +// script records each spawn and then sleeps, and the test binds the local +// port itself to make a spawn "ready". +func TestSuperviseSSHTunnel(t *testing.T) { + sshBinary = "sh" + t.Cleanup(func() { sshBinary = "ssh" }) + + counter := filepath.Join(t.TempDir(), "spawns") + args := []string{"-c", `echo spawn >> "$0"; exec sleep 30`, counter} + port := freePort(t) + + ctx, cancel := context.WithCancel(context.Background()) + t.Cleanup(cancel) + log := zerolog.New(zerolog.NewTestWriter(t)).With().Timestamp().Logger() + ctx = log.WithContext(ctx) + + gate := newTunnelGate() + gate.set(true) + + first, err := startSSHTunnel(ctx, args) + if err != nil { + t.Fatal(err) + } + waitFor(t, 5*time.Second, "first spawn", func() bool { return spawnCount(t, counter) >= 1 }) + + const readyTimeout = 300 * time.Millisecond + supervisorDone := make(chan struct{}) + go func() { + defer close(supervisorDone) + superviseSSHTunnel(ctx, port, args, first, gate, readyTimeout) + }() + + // ssh dies: the gate goes down and a replacement is spawned after the + // 1s minimum backoff. + if err := first.kill(); err != nil { + t.Fatal(err) + } + waitFor(t, 5*time.Second, "gate down", func() bool { return !isUp(gate) }) + waitFor(t, 10*time.Second, "second spawn", func() bool { return spawnCount(t, counter) >= 2 }) + + // Nothing binds the port, so the replacement is killed after readyTimeout + // and a third one is spawned after the 2s backoff. + if isUp(gate) { + t.Fatal("gate came up without anything listening on the port") + } + listener, err := net.Listen("tcp", fmt.Sprintf("127.0.0.1:%d", port)) + if err != nil { + t.Fatal(err) + } + t.Cleanup(func() { + if err := listener.Close(); err != nil { + t.Error(err) + } + }) + waitFor(t, 10*time.Second, "third spawn", func() bool { return spawnCount(t, counter) >= 3 }) + + // The port now accepts, so the third spawn is ready and the gate comes up + // with a new generation. + upCtx, upCancel := context.WithTimeout(ctx, 10*time.Second) + defer upCancel() + gen, err := gate.waitUp(upCtx, 1) + if err != nil { + t.Fatalf("gate never came up: %v", err) + } + if gen != 2 { + t.Fatalf("expected generation 2, got %d", gen) + } + + cancel() + select { + case <-supervisorDone: + case <-time.After(2 * time.Second): + t.Fatal("supervisor did not stop after ctx cancel") + } +} diff --git a/server/server.go b/server/server.go index e137100..4a5fd77 100644 --- a/server/server.go +++ b/server/server.go @@ -75,6 +75,7 @@ type config struct { logging func(ctx context.Context) *zerolog.Logger withoutInitialConnectionCheck bool cookiePath string + interceptors []connect.Interceptor } func newConfig(opts []Option) config { @@ -123,6 +124,14 @@ func WithoutInitialConnectionCheck() Option { } } +// WithInterceptors runs the given interceptors innermost, after logging and +// error mapping, so they see the handler's errors as-is. +func WithInterceptors(interceptors ...connect.Interceptor) Option { + return func(c *config) { + c.interceptors = append(c.interceptors, interceptors...) + } +} + // WithLogging allows you to provide a custom logging function. // // The function will be called with the context of the request, and should @@ -328,6 +337,12 @@ func withCancel[R any, M any]( } } +// ErrUnreachable is the cause of CodeUnavailable when the dial to Bitcoin Core +// is refused. It only comes from a failed dial, and rpcclient sends every +// request on a fresh connection, so a request with this error never reached +// Core and is safe to retry. +var ErrUnreachable = errors.New("unable to connect to Bitcoin Core") + func transformError(ctx context.Context, err error) error { if err == nil { panic("PROGRAMMER ERROR: transformError called with nil") @@ -339,10 +354,7 @@ func transformError(ctx context.Context, err error) error { zerolog.Ctx(ctx).Debug().Err(err). Msgf("transform error: returning connect.CodeUnavailable") - cErr := connect.NewError( - connect.CodeUnavailable, - errors.New("unable to connect to Bitcoin Core"), - ) + cErr := connect.NewError(connect.CodeUnavailable, ErrUnreachable) if detail, detaiLErr := connect.NewErrorDetail( wrapperspb.String(err.Error()), @@ -448,7 +460,7 @@ func (b *Bitcoind) ListUnspent(ctx context.Context, c *connect.Request[pb.ListUn res, err := rpcclient.ReceiveFuture(rpc.SendCmd(ctx, &cmd)) if err != nil { - return nil, err + return nil, transformError(ctx, err) } var parsed []btcjson.ListUnspentResult @@ -478,7 +490,7 @@ func (b *Bitcoind) ListUnspent(ctx context.Context, c *connect.Request[pb.ListUn func (b *Bitcoind) ListWallets(ctx context.Context, _ *connect.Request[emptypb.Empty]) (*connect.Response[pb.ListWalletsResponse], error) { wallets, err := b.rpc.ListWallets(ctx) if err != nil { - return nil, err + return nil, transformError(ctx, err) } return connect.NewResponse(&pb.ListWalletsResponse{ @@ -807,7 +819,7 @@ func (b *Bitcoind) GetBlock(ctx context.Context, c *connect.Request[pb.GetBlockR if c.Msg.Height != nil { hash, err = b.rpc.GetBlockHash(ctx, int64(*c.Msg.Height)) if err != nil { - return nil, fmt.Errorf("get block hash from height: %w", err) + return nil, fmt.Errorf("get block hash from height: %w", transformError(ctx, err)) } } @@ -2542,7 +2554,7 @@ func (b *Bitcoind) setupServer() { b.server = connectserver.New( logging.InterceptorConf{}, - handleBtcJsonErrors(), + append([]connect.Interceptor{handleBtcJsonErrors()}, b.conf.interceptors...)..., ) connectserver.Register(b.server, rpc.NewBitcoinServiceHandler, rpc.BitcoinServiceHandler(b)) }