mirror of
https://github.com/jcreek/CosmicClash.git
synced 2026-09-10 16:04:04 +00:00
Josh Creek
129b0c7ef0
The Deployment runs two replicas, but WebSocket subscribers live only in each process's in-memory hub. Every replica races to read the same global unpublished outbox rows, and publishing succeeded even when the winning replica held no matching local subscriber -- that replica then set the single global published_at. A client connected to the other replica never received the event, and delivery degraded further with each replica added. REST recovery eventually converged, but short-lived proposal transitions could be observed late or not at all. Publish committed events through PostgreSQL LISTEN/NOTIFY so the replica that owns the subscriber's connection delivers it, regardless of which replica drained the row. The listener holds its own pgx connection -- LISTEN is session state, so a pooled database/sql connection cannot carry it -- and reconnects with backoff, since losing it would silently downgrade that replica's subscribers to REST-only recovery. The fan-out is optional: without EventFanout configured, behaviour is unchanged local-hub publication, which stays correct for a single replica and for tests. Only outbox-sourced events are routed through it; the in-request-path publishes remain local, as those are a latency optimisation for the caller's own connection. Fan-out needs a wire shape of its own because ControlPlaneEvent hides PlayerID from clients, and the recipient is exactly what a peer replica needs to route on.
94 lines
3.2 KiB
Go
94 lines
3.2 KiB
Go
package store
|
|
|
|
import (
|
|
"context"
|
|
"database/sql"
|
|
"fmt"
|
|
"time"
|
|
|
|
"github.com/jackc/pgx/v5"
|
|
)
|
|
|
|
// ControlPlaneEventChannel is the PostgreSQL LISTEN/NOTIFY channel used to fan
|
|
// committed outbox events out to every control-plane replica.
|
|
//
|
|
// WebSocket subscribers live in each process's in-memory hub, but the outbox
|
|
// is global: every replica raced to read the same unpublished rows, and the
|
|
// winner set the single global published_at even when it held no matching
|
|
// subscriber. A client connected to any other replica then never received the
|
|
// event, and delivery degraded as replicas were added. Notifying through a
|
|
// shared transport means the replica that owns the connection publishes it,
|
|
// regardless of which replica drained the row.
|
|
const ControlPlaneEventChannel = "cosmic_clash_control_plane_events"
|
|
|
|
// MaxNotifyPayloadBytes is PostgreSQL's hard limit for a NOTIFY payload.
|
|
// Control-plane events are a handful of short fields, so this is a guard
|
|
// against a future field making delivery fail at runtime, not a live concern.
|
|
const MaxNotifyPayloadBytes = 7999
|
|
|
|
// NotifyControlPlaneEvent broadcasts one already-encoded event to every
|
|
// listening replica. It is called after the event's durable commit, so a lost
|
|
// notification degrades to the REST recovery path rather than losing state.
|
|
func NotifyControlPlaneEvent(ctx context.Context, db *sql.DB, payload []byte) error {
|
|
if db == nil || len(payload) == 0 {
|
|
return fmt.Errorf("invalid control-plane event notification")
|
|
}
|
|
if len(payload) > MaxNotifyPayloadBytes {
|
|
return fmt.Errorf("control-plane event payload is %d bytes, over the %d byte NOTIFY limit", len(payload), MaxNotifyPayloadBytes)
|
|
}
|
|
_, err := db.ExecContext(ctx, `SELECT pg_notify($1, $2)`, ControlPlaneEventChannel, string(payload))
|
|
return err
|
|
}
|
|
|
|
// ListenControlPlaneEvents holds a dedicated connection and delivers every
|
|
// notification to handle until ctx is cancelled. It reconnects on failure:
|
|
// losing the listener would silently downgrade this replica's subscribers to
|
|
// REST-only recovery, which is exactly the degradation being fixed.
|
|
//
|
|
// A dedicated pgx connection is required because LISTEN is session state and
|
|
// database/sql may hand any pooled connection to any caller.
|
|
func ListenControlPlaneEvents(ctx context.Context, dsn string, handle func([]byte), onError func(error)) {
|
|
if dsn == "" || handle == nil {
|
|
return
|
|
}
|
|
backoff := time.Second
|
|
for ctx.Err() == nil {
|
|
err := listenOnce(ctx, dsn, handle)
|
|
if ctx.Err() != nil {
|
|
return
|
|
}
|
|
if err != nil && onError != nil {
|
|
onError(err)
|
|
}
|
|
select {
|
|
case <-ctx.Done():
|
|
return
|
|
case <-time.After(backoff):
|
|
}
|
|
if backoff < 30*time.Second {
|
|
backoff *= 2
|
|
}
|
|
}
|
|
}
|
|
|
|
func listenOnce(ctx context.Context, dsn string, handle func([]byte)) error {
|
|
conn, err := pgx.Connect(ctx, dsn)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
defer conn.Close(context.Background())
|
|
if _, err := conn.Exec(ctx, `LISTEN `+pgx.Identifier{ControlPlaneEventChannel}.Sanitize()); err != nil {
|
|
return err
|
|
}
|
|
for {
|
|
notification, err := conn.WaitForNotification(ctx)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
if notification == nil || notification.Payload == "" {
|
|
continue
|
|
}
|
|
handle([]byte(notification.Payload))
|
|
}
|
|
}
|