Correlation ID echo for async WebSocket session responses — prevents routing collisions on concurrent commands

session-response-correlation-id.go

// Correlation ID echo for async WebSocket responses.
//
// Context: a single WebSocket per instance carries two kinds of commands
// (user queries + background health polls). Without a request-level ID,
// whichever waiter grabbed the response first "won" and the other timed out.
//
// The fix was one field: correlationId, generated by the hub, echoed verbatim
// in every response. The hub routes responses to the correct waiter by that
// field, never by sessionId.
//
// Blog post: https://www.linkedin.com/in/fernando-possebon/

package uplink

import "strings"

// sessResp builds a session_response base map with the required routing fields.
// If correlationID is non-empty it is included so the hub can match the response
// to the originating request (prevents collisions from concurrent commands).
func sessResp(sessionID, correlationID string) map[string]interface{} {
	m := map[string]interface{}{
		"type":      "session_response",
		"sessionId": sessionID,
	}
	if correlationID != "" {
		m["correlationId"] = correlationID
	}
	return m
}

// Usage inside handleSessionCommand — the correlationId is read from the
// incoming message and echoed in EVERY response path (error, block, success).
// sessResp is the single choke point. There is no other way to build a response.
// That's what makes "echo correlationId" a one-line invariant instead of a
// scavenger hunt across handlers.
func (u *Uplink) handleSessionCommand(msg map[string]interface{}) {
	sessionID, _ := msg["sessionId"].(string)
	if sessionID == "" {
		sessionID, _ = msg["session_id"].(string) // legacy fallback
	}
	correlationID, _ := msg["correlationId"].(string)
	query, _ := msg["query"].(string)

	if strings.TrimSpace(query) == "" {
		resp := sessResp(sessionID, correlationID)
		resp["error"] = "empty query"
		u.writeJSON(resp)
		return
	}

	// ... resolve collector, run the query, build resp the same way.
	// Every response path ALWAYS goes through sessResp(sessionID, correlationID).
}

// Design notes for the next incident review:
//
//  1. correlationId is read but not required. Older hubs that don't send it get
//     legacy routing. No flag day, no breaking migration.
//
//  2. sessResp is the only response-envelope builder in the package. You cannot
//     accidentally forget to echo correlationId because there is no other path.
//
//  3. The hub generates the IDs; the agent never does. Lets us trace a request
//     end-to-end from the UI click through the WebSocket through the database
//     and back, joining on one field.
//
// Lesson paid for in timeouts: any async messaging layer needs a request ID
// before you think it does. Session/connection IDs route messages to the right
// process. They don't route responses to the right caller. Those are different
// problems. HTTP pretends they are the same because request-response is
// synchronous. The moment you go async, the illusion costs you.