package upstream import ( "context" "errors" "testing" ) func TestHandleConnectionLossNotifiesPendingSubmitters(t *testing.T) { var reported ConnectionState conn := &connection{ pending: make(map[uint32]chan submitPartResponse), pool: &connectionPool{ channelID: "channel-1", connectionID: "channel-1:primary", config: normalizeUpstreamConfig(queueUpstreamConfigForTest()), reporter: func(_ context.Context, state ConnectionState) error { reported = state return nil }, }, } conn.pool.conns = []*connection{conn} waiter := make(chan submitPartResponse, 1) conn.pending[7] = waiter loss := errors.New("socket closed") conn.handleConnectionLoss(loss) select { case result := <-waiter: if !errors.Is(result.err, loss) { t.Fatalf("pending waiter err = %v, want %v", result.err, loss) } default: t.Fatal("expected pending waiter to be notified") } if !conn.closed { t.Fatal("expected connection to be marked closed") } if len(conn.pending) != 0 { t.Fatalf("expected pending map to be reset, got %d entries", len(conn.pending)) } if reported.Status != "disconnected" || reported.CurrentConnections != 0 { t.Fatalf("unexpected reported state: %+v", reported) } } func TestTemporaryReadTimeoutDetection(t *testing.T) { if !isTemporaryReadTimeout(fakeNetError{timeout: true}) { t.Fatal("expected timeout error to be treated as temporary") } if isTemporaryReadTimeout(errors.New("eof")) { t.Fatal("did not expect non-timeout error to be treated as temporary") } } type fakeNetError struct { timeout bool } func (f fakeNetError) Error() string { return "network error" } func (f fakeNetError) Timeout() bool { return f.timeout } func (f fakeNetError) Temporary() bool { return f.timeout }