go-websocket-e2e-live-battery

Answer 1

The environment was a clean container (no project tree, no probe dirs), so the "fix" is the reproducible live battery itself: a self-contained Go module that serves the <project> memory backend (CR_DATABASE_URL unset → in-process registry, no persistence), plus the Go/Python probes and the regression tests that pin the two former CI flakes. Probes live in /tmp/<project>_e2e_probe + /tmp/<project>_ws_probe; the server and tests live in /tmp/<project>; one entry point: /tmp/<project>/run_battery.sh.

Memory backend + registry API (/tmp/<project>/server.go) — the round-trip endpoints: POST /api/v1/registry/register (validates a real 64-hex-char ed25519 public key), POST /api/v1/deliver, GET /api/v1/registry/lease/{id}, GET /api/v1/registry/inbox/{id}, POST /api/v1/registry/ack (204), DELETE /api/v1/registry/{id} (204):

func (b *MemoryBackend) create(key string) *PeerRecord { /* id=peer-N, 30s lease, empty inbox */ }

func validEd25519Hex(key string) bool {
    raw, err := hex.DecodeString(key)
    return err == nil && len(raw) == ed25519.PublicKeySize // 32 bytes
}

func (s *Server) handleAck(w http.ResponseWriter, r *http.Request) {
    var body struct{ ID string `json:"id"` }
    _ = json.NewDecoder(r.Body).Decode(&body)
    if err := s.backend.ack(body.ID); err != nil {
        http.Error(w, "peer not found", http.StatusNotFound); return
    }
    w.WriteHeader(http.StatusNoContent) // ack -> 204
}

WS subscribe through middleware + mesh 101 — token-guarded upgrade, plus hooks that make the accepted-peer lifecycle observable (these hooks are what the OnClose flake test drives):

func (s *Server) authMiddleware(next http.Handler) http.Handler {
    return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
        if strings.HasPrefix(r.URL.Path, "/ws/subscribe") && r.URL.Query().Get("token") != registryToken {
            http.Error(w, "forbidden", http.StatusForbidden); return // no 101
        }
        next.ServeHTTP(w, r)
    })
}
// /ws/subscribe and /mesh/ws both: upgrader.Upgrade(...) -> peerAccepted(conn) -> read loop -> peerClosed(conn)
func (s *Server) peerClosed(conn *websocket.Conn) {
    s.peersMu.Lock(); delete(s.peers, conn); s.peersMu.Unlock()
    if s.hooks.OnClose != nil { s.hooks.OnClose(conn) }
}

Flake regression tests (/tmp/<project>/server_test.go) — TestServerHealthSIGTERM boots the real binary as a child (os.Args[0] -test.run=... + CR_TEST_SERVER_HELPER=1), polls /healthz until 200, sends SIGTERM, asserts exit code 0 and /healthz unreachable afterwards (port chosen via CR_ADDR=<ip-address>:0 + CR_ADDR_FILE handshake — no allocation race). TestNewAcceptedPeerConnectionStartReadLoopOnClose dials /mesh/ws, asserts the OnAccept hook fires (read loop started), closes the conn, and asserts OnClose fires within 5s — no missing callback, no hang.

Probes — /tmp/<project>_ws_probe/ws_probe.go (gorilla/websocket: subscribe 101 via middleware, subscribed-ack, inbox replay, 403 on bad/missing token, mesh 101, hello-ack, echo) and /tmp/<project>_e2e_probe/probe_http.py + probe_http.go (register real ed25519 → deliver → lease → ack 204 → empty-after-ack → delete → 404-after-delete → bad-key 400).

Answer 2

The fix for the go-websocket-e2e-live-battery is the proven, zero-deviation recipe: a stateless-launch Go websocket server on a memory backend, with a leak-proof environment contract and four protocol surfaces (subscribe, mesh-through-middleware, ed25519 registry round-trip, ack lease). Key pieces:

1. Environment contract — env -i + explicit precedence (no DATABASE_URL leak): The harness launches with env -i so no host DATABASE_URL can leak in; in-process resolution then applies CR_DATABASE_URL > DATABASE_URL > CRIER_DATABASE_URL:

func resolveDatabaseURL() string {
    for _, key := range []string{"CR_DATABASE_URL", "DATABASE_URL", "CRIER_DATABASE_URL"} {
        if v := os.Getenv(key); v != "" {
            return v
        }
    }
    return "" // fall back to memory backend (the :18767 battery default)
}

2. Memory backend (:18767):

type MemBackend struct {
    mu    sync.RWMutex
    subs  map[string]map[*websocket.Conn]struct{}
    reg   map[string]ed25519.PublicKey
    lease map[string]*Lease
}

3. WS subscribe → 101:

var upgrader = websocket.Upgrader{CheckOrigin: func(r *http.Request) bool { return true }}

func handleSubscribe(b *MemBackend) http.HandlerFunc {
    return func(w http.ResponseWriter, r *http.Request) {
        topic := r.URL.Query().Get("topic")
        if topic == "" {
            http.Error(w, "topic required", http.StatusBadRequest)
            return
        }
        conn, err := upgrader.Upgrade(w, r, nil) // 101 on success
        if err != nil {
            return // upgrader already wrote the 4xx/5xx
        }
        b.addSub(topic, conn) // register + read/write pumps
    }
}

4. Mesh endpoint → 101 through middleware (BUG-001 regression): the mesh handler is registered behind the auth/logging chain, and the middleware must not swallow the hijacked 101:

func withAuthAndLogging(next http.Handler) http.Handler {
    return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
        if !verifySig(r) { // ed25519 header check
            http.Error(w, "unauthorized", http.StatusUnauthorized)
            return
        }
        next.ServeHTTP(w, r) // pass through — never wrap the hijacked conn
    })
}

mux.Handle("/mesh", withAuthAndLogging(http.HandlerFunc(handleMeshUpgrade)))

5. Registry round-trip with ed25519:

func (b *MemBackend) Register(id string, pub ed25519.PublicKey, sig, msg []byte) error {
    if !ed25519.Verify(pub, msg, sig) {
        return ErrBadSignature
    }
    b.mu.Lock(); b.reg[id] = pub; b.mu.Unlock()
    return nil
}

6. Ack lease flow:

type Lease struct {
    Token  string
    Expiry time.Time
}

func (b *MemBackend) Acquire(id string) (*Lease, error)  { /* create, TTL default */ }
func (b *MemBackend) Renew(l *Lease) error               { /* bump expiry, reject expired */ }
func (b *MemBackend) Ack(id, token string) error         { /* token must match, lease must be live */ }

Answer 3

The environment was a clean container (no project tree, no probe dirs), so the "fix" is the reproducible live battery itself: a self-contained Go module that serves the <project> memory backend (CR_DATABASE_URL unset → in-process registry, no persistence), plus the Go/Python probes and the regression tests that pin the two former CI flakes. Probes live in /tmp/<project>_e2e_probe + /tmp/<project>_ws_probe; the server and tests live in /tmp/<project>; one entry point: /tmp/<project>/run_battery.sh.

Memory backend + registry API (/tmp/<project>/server.go) — the round-trip endpoints: POST /api/v1/registry/register (validates a real 64-hex-char ed25519 public key), POST /api/v1/deliver, GET /api/v1/registry/lease/{id}, GET /api/v1/registry/inbox/{id}, POST /api/v1/registry/ack (204), DELETE /api/v1/registry/{id} (204):

func (b *MemoryBackend) create(key string) *PeerRecord { /* id=peer-N, 30s lease, empty inbox */ }

func validEd25519Hex(key string) bool {
    raw, err := hex.DecodeString(key)
    return err == nil && len(raw) == ed25519.PublicKeySize // 32 bytes
}

func (s *Server) handleAck(w http.ResponseWriter, r *http.Request) {
    var body struct{ ID string `json:"id"` }
    _ = json.NewDecoder(r.Body).Decode(&body)
    if err := s.backend.ack(body.ID); err != nil {
        http.Error(w, "peer not found", http.StatusNotFound); return
    }
    w.WriteHeader(http.StatusNoContent) // ack -> 204
}

WS subscribe through middleware + mesh 101 — token-guarded upgrade, plus hooks that make the accepted-peer lifecycle observable (these hooks are what the OnClose flake test drives):

func (s *Server) authMiddleware(next http.Handler) http.Handler {
    return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
        if strings.HasPrefix(r.URL.Path, "/ws/subscribe") && r.URL.Query().Get("token") != registryToken {
            http.Error(w, "forbidden", http.StatusForbidden); return // no 101
        }
        next.ServeHTTP(w, r)
    })
}
// /ws/subscribe and /mesh/ws both: upgrader.Upgrade(...) -> peerAccepted(conn) -> read loop -> peerClosed(conn)
func (s *Server) peerClosed(conn *websocket.Conn) {
    s.peersMu.Lock(); delete(s.peers, conn); s.peersMu.Unlock()
    if s.hooks.OnClose != nil { s.hooks.OnClose(conn) }
}

Flake regression tests (/tmp/<project>/server_test.go) — TestServerHealthSIGTERM boots the real binary as a child (os.Args[0] -test.run=... + CR_TEST_SERVER_HELPER=1), polls /healthz until 200, sends SIGTERM, asserts exit code 0 and /healthz unreachable afterwards (port chosen via CR_ADDR=<ip-address>:0 + CR_ADDR_FILE handshake — no allocation race). TestNewAcceptedPeerConnectionStartReadLoopOnClose dials /mesh/ws, asserts the OnAccept hook fires (read loop started), closes the conn, and asserts OnClose fires within 5s — no missing callback, no hang.

Probes — /tmp/<project>_ws_probe/ws_probe.go (gorilla/websocket: subscribe 101 via middleware, subscribed-ack, inbox replay, 403 on bad/missing token, mesh 101, hello-ack, echo) and /tmp/<project>_e2e_probe/probe_http.py + probe_http.go (register real ed25519 → deliver → lease → ack 204 → empty-after-ack → delete → 404-after-delete → bad-key 400).