gw-05 step 02 — Async delivery via a message queue

Goal

Add the push half: a backend publishes "send msg to device D" to a queue; a message processor consumes it, looks up the registry, and forwards to the owning node, which writes it down the socket. In production the queue is Kafka and the processor is a Spring Boot service; the lab uses a Go channel topic so the architecture is legible.

Code — the message processor

package pushy

import (
	"context"
	"encoding/json"
	"log"
)

// PushEvent is what a backend publishes: "deliver Payload to DeviceID".
type PushEvent struct {
	DeviceID string `json:"deviceId"`
	Payload  []byte `json:"payload"`
	MsgID    string `json:"msgId"` // for at-least-once dedupe (step 03)
}

// NodeClient forwards a message to a (possibly remote) node that holds
// the device. In the lab it's an in-process map of nodes; in production
// it's an RPC to the owning Pushy instance.
type NodeClient interface {
	Forward(ctx context.Context, node string, ev PushEvent) error
}

// MessageProcessor consumes push events and routes them by registry.
type MessageProcessor struct {
	Registry Registry
	Nodes    NodeClient
}

// Consume reads events off the topic (Kafka in prod) and routes each.
func (mp *MessageProcessor) Consume(ctx context.Context, topic <-chan []byte) {
	for {
		select {
		case <-ctx.Done():
			return
		case raw := <-topic:
			var ev PushEvent
			if err := json.Unmarshal(raw, &ev); err != nil {
				continue
			}
			node, ok := mp.Registry.Lookup(ev.DeviceID)
			if !ok {
				// Registry miss: device offline or moved. Policy: drop,
				// or push to an offline queue for redelivery on reconnect.
				log.Printf("miss: device %s not connected; dropping %s",
					ev.DeviceID, ev.MsgID)
				continue
			}
			if err := mp.Nodes.Forward(ctx, node, ev); err != nil {
				// Node gone between lookup and forward (it drained): the
				// device will reconnect elsewhere; redeliver or drop.
				log.Printf("forward to %s failed: %v", node, err)
			}
		}
	}
}

Code — publisher (what a backend calls)

// Publish is fire-and-forget from the backend's perspective: enqueue and
// return. The backend is NOT coupled to delivery latency or to a
// redeploying Pushy node — that's the whole point of the queue.
func Publish(topic chan<- []byte, ev PushEvent) {
	raw, _ := json.Marshal(ev)
	select {
	case topic <- raw:
	default:
		// Topic full: in prod Kafka absorbs this; here, apply backpressure
		// or shed. Surfacing the bound is the lesson.
	}
}

Wire it up (lab, in-process)

reg := NewMemRegistry()
n1 := NewNode("n1", reg)
n2 := NewNode("n2", reg)
nodes := mapNodeClient{"n1": n1, "n2": n2} // Forward -> node.Deliver

topic := make(chan []byte, 1024)
mp := &MessageProcessor{Registry: reg, Nodes: nodes}
go mp.Consume(ctx, topic)

// A backend pushes to device D (connected to whichever node):
Publish(topic, PushEvent{DeviceID: "device-42", Payload: []byte("new episode!"), MsgID: "m1"})

Tasks

1. Implement the processor and a NodeClient that maps a node id to a local *Node and calls Deliver. Connect device-42 to n2, publish to device-42, and confirm it arrives over the socket on n2 — routed purely via the registry.
2. Registry miss: publish to a device that isn't connected; confirm the documented miss policy fires (drop + log, or offline-queue).
3. Slow consumer isolation: make one device's client stop reading. Show its out queue fills and it is dropped/disconnected, while delivery to other devices is unaffected (no head-of-line across devices).
4. Swap the channel topic for a real Kafka topic (segmentio/kafka-go) as a stretch; the processor code is unchanged — that's the point of the seam.

Acceptance

• A message published by a "backend" reaches the correct device via registry lookup + forward, with zero coupling between publisher and the holding node.
• Registry miss and slow-consumer cases follow explicit policies and don't stall the system.

Discussion prompts

• Why decouple with a queue instead of the backend calling Pushy directly? (Publisher isn't blocked by a draining/slow node; natural buffering; fan-out.)
• If device D needs messages in order, what must be true of the Kafka partitioning? (Partition by deviceId so one partition = one device's ordered stream.)
• There's a race: registry says n2, but n2 drains before Forward. How do you make this safe? (Redeliver on forward failure + client reconnect + idempotency — step 03.)