fix: close 16 defects from adversarial bug-hunt sweep

cmd/bodek/upgrade.go

@@ -8,6 +8,7 @@ import (
 	"context"
 	"crypto/sha256"
 	"encoding/hex"
+	"errors"
 	"fmt"
 	"io"
 	"net/http"
@@ -101,15 +102,27 @@ func archiveName(tag, goos, goarch string) string {
 	return fmt.Sprintf("bodek_%s_%s_%s.%s", tag, goos, goarch, ext)
 }
 
+// downloadTimeout bounds a single release-asset body read. It must be a
+// per-request deadline, not Client.Timeout: the timeout bounds the WHOLE
+// transfer including the body, so a multi-MB archive over a slow link would
+// fail every time under the short API budget.
+const downloadTimeout = 10 * time.Minute
+
 // download fetches url into memory. Release archives are a few MB, so a
 // buffered read is fine and keeps checksum verification straightforward.
 func download(ctx context.Context, client *http.Client, url string) ([]byte, error) {
+	// A copy without the overall Timeout: the deadline below bounds the
+	// read instead. The transport (TLS cache, dialer) is shared.
+	dl := *client
+	dl.Timeout = 0
+	ctx, cancel := context.WithTimeout(ctx, downloadTimeout)
+	defer cancel()
 	req, err := http.NewRequestWithContext(ctx, http.MethodGet, url, nil)
 	if err != nil {
 		return nil, fmt.Errorf("build download request: %w", err)
 	}
 	req.Header.Set("User-Agent", "bodek-updater")
-	resp, err := client.Do(req)
+	resp, err := dl.Do(req)
 	if err != nil {
 		return nil, fmt.Errorf("download %s: %w", url, err)
 	}
@@ -210,10 +223,18 @@ func extractZip(archive []byte) ([]byte, error) {
 	return nil, fmt.Errorf("archive contains no bodek binary")
 }
 
+// renameFn and syncFile are test hooks over os.Rename and (*os.File).Sync
+// so the swap sequence is injectable.
+var (
+	renameFn = os.Rename
+	syncFile = (*os.File).Sync
+)
+
 // replaceExecutable atomically swaps the binary at target with data: the new
-// file is written next to the target and renamed over it, so a crash
-// mid-upgrade never leaves a truncated binary. target is resolved through
-// symlinks first so `go install` shims and PATH links are not clobbered.
+// file is written next to the target, fsynced, and renamed over the old one,
+// so a crash mid-upgrade never leaves a truncated binary. target is resolved
+// through symlinks first so `go install` shims and PATH links are not
+// clobbered.
 func replaceExecutable(data []byte, target string) error {
 	resolved, err := filepath.EvalSymlinks(target)
 	if err != nil {
@@ -224,8 +245,15 @@ func replaceExecutable(data []byte, target string) error {
 		return fmt.Errorf("create temp file next to %s: %w", resolved, err)
 	}
 	tmpName := tmp.Name()
-	// No-op once the rename below has moved the temp file into place.
-	defer func() { _ = os.Remove(tmpName) }()
+	// No-op once the rename below has moved the temp file into place — but
+	// only the success path: the Windows rollback-fail path leaves the new
+	// binary at tmpName on purpose, so installFailed keeps it alive.
+	installFailed := false
+	defer func() {
+		if !installFailed {
+			_ = os.Remove(tmpName)
+		}
+	}()
 	if _, err := tmp.Write(data); err != nil {
 		_ = tmp.Close()
 		return fmt.Errorf("write new binary: %w", err)
@@ -237,21 +265,54 @@ func replaceExecutable(data []byte, target string) error {
 	if err := tmp.Close(); err != nil {
 		return fmt.Errorf("flush new binary: %w", err)
 	}
-	if err := os.Rename(tmpName, resolved); err != nil {
+	reopened, err := os.Open(tmpName)
+	if err == nil {
+		err = syncFile(reopened)
+		_ = reopened.Close()
+	}
+	if err != nil {
+		return fmt.Errorf("sync new binary: %w", err)
+	}
+	if err := renameFn(tmpName, resolved); err != nil {
 		if runtime.GOOS != "windows" {
 			return fmt.Errorf("replace %s: %w", resolved, err)
 		}
-		// Windows refuses to rename over a running executable; move the old
-		// one aside first, then drop the new binary into place.
-		old := resolved + ".old"
-		_ = os.Remove(old)
-		if rerr := os.Rename(resolved, old); rerr != nil {
-			return fmt.Errorf("move current executable aside: %w", rerr)
+		// Windows refuses to rename over a running executable: swap the old
+		// one aside, then drop the new binary into place (with rollback).
+		if serr := swapAsideWindows(resolved, tmpName); serr != nil {
+			// On the double-failure path the new binary deliberately
+			// survives at tmpName — keep the deferred cleanup off it.
+			var kept tmpKeptError
+			installFailed = errors.As(serr, &kept)
+			return serr
 		}
-		if rerr := os.Rename(tmpName, resolved); rerr != nil {
-			return fmt.Errorf("install new binary: %w", rerr)
+		return nil
+	}
+	return nil
+}
+
+// tmpKeptError marks a failure where the staged new binary intentionally
+// survives at its temp path (both Windows renames failed) — the deferred
+// cleanup in replaceExecutable must not delete it.
+type tmpKeptError struct{ error }
+
+// swapAsideWindows installs tmpName over resolved on Windows, moving the
+// running binary to resolved+".old" first. A failed install rename rolls
+// the old binary back — the swap must never strand the executable as .old.
+func swapAsideWindows(resolved, tmpName string) error {
+	old := resolved + ".old"
+	_ = os.Remove(old)
+	if rerr := renameFn(resolved, old); rerr != nil {
+		return fmt.Errorf("move current executable aside: %w", rerr)
+	}
+	if rerr := renameFn(tmpName, resolved); rerr != nil {
+		if rberr := renameFn(old, resolved); rberr != nil {
+			// Both renames failed: the new binary stays at tmpName on
+			// purpose — tell the operator where both halves live.
+			return tmpKeptError{fmt.Errorf("install new binary: %w (rollback also failed: %v — old binary is at %s, new binary is at %s)", rerr, rberr, old, tmpName)}
 		}
-		_ = os.Remove(old) // best effort: a locked .old goes away on a later run
+		return fmt.Errorf("install new binary: %w", rerr)
 	}
+	_ = os.Remove(old) // best effort: a locked .old goes away on a later run
 	return nil
 }

cmd/bodek/upgrade_wave3_test.go

@@ -0,0 +1,117 @@
+package main
+
+import (
+	"context"
+	"errors"
+	"net/http"
+	"net/http/httptest"
+	"os"
+	"path/filepath"
+	"testing"
+	"time"
+
+	"github.com/BackendStack21/bodek/internal/update"
+)
+
+// TestSwapAsideWindowsRollback guards the Windows swap: if the second
+// rename (new binary into place) fails after the running binary was moved
+// aside, the old binary must be restored — otherwise the install path is
+// bricked with the executable stranded as .old.
+func TestSwapAsideWindowsRollback(t *testing.T) {
+	dir := t.TempDir()
+	target := filepath.Join(dir, "bodek")
+	if err := os.WriteFile(target, []byte("old-binary"), 0o755); err != nil {
+		t.Fatal(err)
+	}
+	tmp := filepath.Join(dir, ".bodek-upgrade-new")
+	if err := os.WriteFile(tmp, []byte("new-binary"), 0o755); err != nil {
+		t.Fatal(err)
+	}
+
+	realRename := renameFn
+	defer func() { renameFn = realRename }()
+	calls := 0
+	renameFn = func(from, to string) error {
+		calls++
+		if calls == 2 { // the "install new binary" rename fails
+			return errors.New("access denied")
+		}
+		return realRename(from, to)
+	}
+
+	err := swapAsideWindows(target, tmp)
+	if err == nil {
+		t.Fatal("expected the injected rename failure to surface")
+	}
+	got, rerr := os.ReadFile(target)
+	if rerr != nil {
+		t.Fatalf("executable missing after failed swap: %v", rerr)
+	}
+	if string(got) != "old-binary" {
+		t.Fatalf("old binary not restored after failed swap: %q", got)
+	}
+}
+
+// TestReplaceExecutableSyncsBeforeRename guards durability: the new binary
+// must be fsynced before the rename, or a crash can persist the rename
+// with no/partial data behind it — a zero-length bodek — despite the
+// "never leaves a truncated binary" contract.
+func TestReplaceExecutableSyncsBeforeRename(t *testing.T) {
+	dir := t.TempDir()
+	target := filepath.Join(dir, "bodek")
+	if err := os.WriteFile(target, []byte("old"), 0o755); err != nil {
+		t.Fatal(err)
+	}
+
+	realSync := syncFile
+	defer func() { syncFile = realSync }()
+	synced := false
+	syncFile = func(f *os.File) error { synced = true; return realSync(f) }
+
+	if err := replaceExecutable([]byte("new"), target); err != nil {
+		t.Fatalf("replaceExecutable: %v", err)
+	}
+	if !synced {
+		t.Fatal("new binary was never fsynced before the rename")
+	}
+}
+
+// TestUpgradeSlowLinkDownloads guards the download budget: Client.Timeout
+// bounds the whole body read, so reusing the short API client for a
+// multi-MB archive fails every transfer slower than that budget. The
+// download path must not inherit the API client's overall deadline.
+func TestUpgradeSlowLinkDownloads(t *testing.T) {
+	archive := buildTarGz(t, "bodek", []byte("slow-payload"))
+	slow := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		time.Sleep(2 * time.Second)
+		_, _ = w.Write(archive)
+	}))
+	defer slow.Close()
+
+	// A 300ms overall client budget vs a 2s drip — margins wide enough to
+	// be CI-stable in both directions.
+	c := &http.Client{Timeout: 300 * time.Millisecond}
+	data, err := download(context.Background(), c, slow.URL+"/bodek.tar.gz")
+	if err != nil {
+		t.Fatalf("download on a slow link failed: %v", err)
+	}
+	if len(data) == 0 {
+		t.Fatal("empty download")
+	}
+}
+
+// TestNewerPseudoVersion guards commit-installed builds: a Go pseudo-version
+// stamp (v0.1.3-0.20260901abcdef12-abc1234) must compare by its release
+// prefix, not report "already up to date" against every future release.
+func TestNewerPseudoVersion(t *testing.T) {
+	if !update.Newer("v9.9.9", "v0.1.3-0.20260901000000-abc1234") {
+		t.Fatal("v9.9.9 must be newer than a v0.1.3 pseudo-version stamp")
+	}
+	if update.Newer("v0.1.3", "v0.1.3-0.20260901000000-abc1234") {
+		t.Fatal("v0.1.3 must not upgrade over a v0.1.3 pseudo-version")
+	}
+	// A genuine semver prerelease must not be misparsed as a pseudo-version.
+	if !update.Newer("v1.3.0", "v1.2.0-0.1") {
+		t.Fatal("v1.3.0 must be newer than prerelease v1.2.0-0.1")
+	}
+}

internal/client/client.go

@@ -248,6 +248,12 @@ type Client struct {
 	stopOnce   sync.Once    // lazy: only jobs-stop pays the longer timeout budget
 	stopClient *http.Client // 12s — odek's stop endpoint blocks stopGrace+4s
 	slowHTTP   *http.Client // 30s — bulk payload reads (session detail, export)
+
+	// done is closed by Close. Every Events send selects on it: when a
+	// reconnect abandons a full Events channel, readLoop must exit instead of
+	// parking forever on a send nobody will ever receive.
+	done      chan struct{}
+	closeOnce sync.Once
 }
 
 // Dial connects to an odek serve WebSocket. wsURL is the ws:// endpoint,
@@ -281,6 +287,7 @@ func Dial(wsURL, origin, baseURL, token string) (*Client, error) {
 		// dedicated 30s client serves them so the 3s interactive budget does
 		// not cut their body reads.
 		slowHTTP: &http.Client{Timeout: 30 * time.Second},
+		done:     make(chan struct{}),
 	}
 	go c.readLoop()
 	return c, nil
@@ -363,7 +370,7 @@ func (c *Client) readLoop() {
 		if pending == nil {
 			return
 		}
-		c.Events <- *pending
+		c.emit(*pending)
 		pending = nil
 		n = 0
 	}
@@ -372,7 +379,7 @@ func (c *Client) readLoop() {
 		_ = c.conn.SetReadDeadline(time.Now().Add(readIdleTimeout))
 		if err := ws.Message.Receive(c.conn, &data); err != nil {
 			flush()
-			c.Events <- Event{Type: EventDisconnected}
+			c.emit(Event{Type: EventDisconnected})
 			return
 		}
 		_ = c.conn.SetReadDeadline(time.Time{}) // received: drop the deadline while decoding
@@ -397,7 +404,7 @@ func (c *Client) readLoop() {
 			continue
 		}
 		flush()
-		c.Events <- ev
+		c.emit(ev)
 	}
 }
 
@@ -524,10 +531,24 @@ func (c *Client) send(v any) error {
 	return ws.JSON.Send(c.conn, v)
 }
 
-// Close shuts the connection.
+// Close shuts the connection and releases a readLoop parked on Events.
 func (c *Client) Close() error {
+	c.closeOnce.Do(func() {
+		if c.done != nil {
+			close(c.done)
+		}
+	})
 	if c.conn == nil {
 		return nil
 	}
 	return c.conn.Close()
 }
+
+// emit delivers ev to Events unless the client is closed. A send to an
+// abandoned full channel would otherwise park readLoop forever.
+func (c *Client) emit(ev Event) {
+	select {
+	case c.Events <- ev:
+	case <-c.done:
+	}
+}

internal/client/reconnect_leak_test.go

@@ -0,0 +1,62 @@
+package client
+
+import (
+	"net/http"
+	"net/http/httptest"
+	"runtime"
+	"testing"
+	"time"
+
+	ws "golang.org/x/net/websocket"
+)
+
+// TestReadLoopExitsWhenEventsAbandoned guards the reconnect swap: when the
+// consumer abandons a full Events channel (a reconnect during a delta
+// firehose), closing the socket must let readLoop exit instead of parking
+// forever on a send to a channel nobody reads.
+func TestReadLoopExitsWhenEventsAbandoned(t *testing.T) {
+	// A 10x flood guarantees the parked state: once the consumer stops
+	// draining, readLoop refills the channel and parks on a send.
+	flood := eventBuffer * 10
+	done := make(chan struct{})
+	mux := http.NewServeMux()
+	mux.Handle("/ws", ws.Handler(func(c *ws.Conn) {
+		defer close(done)
+		for i := 0; i < flood; i++ {
+			if err := ws.Message.Send(c, `{"type":"note","content":"x"}`); err != nil {
+				return
+			}
+		}
+	}))
+	srv := httptest.NewServer(mux)
+	defer srv.Close()
+	wsURL := "ws" + srv.URL[len("http"):]
+	base := runtime.NumGoroutine() // before Dial: excludes readLoop entirely
+	cl, err := Dial(wsURL+"/ws", srv.URL, srv.URL, "test-token")
+	if err != nil {
+		t.Fatalf("Dial: %v", err)
+	}
+
+	// Consume exactly the channel capacity, then vanish — the readLoop is
+	// provably parked on a send to the full channel, exactly the reconnect
+	// swap's state when reconnect.go drops the old client mid-firehose.
+	for i := 0; i < eventBuffer; i++ {
+		select {
+		case <-cl.Events:
+		case <-time.After(10 * time.Second):
+			t.Fatalf("stalled draining event %d", i)
+		}
+	}
+	// The 104 overflow frames are in flight; give readLoop a moment to park.
+	time.Sleep(200 * time.Millisecond)
+
+	_ = cl.Close()
+
+	deadline := time.Now().Add(5 * time.Second)
+	for runtime.NumGoroutine() > base && time.Now().Before(deadline) {
+		time.Sleep(50 * time.Millisecond)
+	}
+	if n := runtime.NumGoroutine(); n > base {
+		t.Fatalf("readLoop goroutine leaked after Close on an abandoned full Events channel: %d goroutines (base %d)", n, base)
+	}
+}