14b65389ef
Tests · UI / test (push) Successful in 2m35s
Tests · Go / test (push) Successful in 1m56s
Tests · UI / test (pull_request) Has been cancelled
Tests · Integration / integration (pull_request) Successful in 1m42s
Tests · Go / test (pull_request) Successful in 2m0s
Browser fetch-streaming layers close response bodies they consider
idle after roughly 15-30 s without incoming bytes. Safari is the
most aggressive, but the symptom matters everywhere: a quiet
SubscribeEvents stream (lobby, between turns, mailbox empty) gets
torn down by the browser, the EventStream singleton reconnects with
backoff, and any push event that fires inside the reconnect window
is lost because `push.Hub` queues are not persisted across
subscription closes. The user-visible failure mode is the
intermittent "Fetch API cannot load … due to access control checks"
console error (a misleading WebKit symptom — CORS headers are
actually present) plus missed turn-ready / mail-received toasts.
Server-side fix: a silence-based heartbeat at the
`authenticatedPushStreamService` wrapper layer. After the signed
`gateway.server_time` bootstrap event, gateway wraps the bound
stream with `heartbeatingStream`. Every tail Send (fan-out, future
variants) resets the silence timer; when the timer elapses, a
goroutine emits `gateway.heartbeat` with only `EventType` set —
everything else stays at proto3 defaults, so the wire frame is
~45 bytes amortised. A `sendMu` serialises the heartbeat goroutine
with tail Sends because grpc.ServerStream.Send is not goroutine-safe.
The heartbeat is intentionally UNSIGNED: heartbeats carry no
payload, dispatch to no handler on the client, and an injected
heartbeat trivially causes no user-visible state change. TLS still
protects the wire and real events keep the signed envelope
unchanged. Documented in `docs/ARCHITECTURE.md` § 15 alongside the
per-scale bandwidth projection (100…100 000 clients × 15…60 s).
Config: new `GATEWAY_PUSH_HEARTBEAT_INTERVAL` (default `15s`,
`0s` disables). Telemetry: new
`gateway.push.heartbeats_sent{outcome}` counter so operators can
budget bandwidth and spot a sudden `outcome=error` bump as an
upstream-failing-before-flush signal.
Client (`ui/frontend/src/api/events.svelte.ts`): early `continue`
on `event.eventType === "gateway.heartbeat"` before `verifyEvent`,
`verifyPayloadHash`, or dispatch — empty signature would otherwise
trip SignatureError and reconnect. A leading heartbeat still flips
`connectionStatus` to `connected` and resets backoff, because
receiving one is proof the stream is healthy.
Tests:
- `push_heartbeat_test.go`: unit tests for the wrapper — zero
interval returns nil, heartbeat fires after silence, real Send
resets the timer, Stop / context-cancel halt the goroutine,
Send errors propagate.
- `server_test.go`: integration tests through the full gateway
pipeline — heartbeat fires after the configured silence window,
zero interval keeps the stream silent.
- `config_test.go`: default applied, env-override parsed,
negative value rejected.
- `events.test.ts`: heartbeat skipped before verification + not
dispatched to handlers; leading heartbeat still flips
`connectionStatus` to `connected`.
Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
277 lines
8.5 KiB
Go
277 lines
8.5 KiB
Go
package config
|
|
|
|
import (
|
|
"crypto/ed25519"
|
|
"crypto/sha256"
|
|
"crypto/x509"
|
|
"encoding/pem"
|
|
"os"
|
|
"path/filepath"
|
|
"sync"
|
|
"testing"
|
|
"time"
|
|
|
|
"galaxy/redisconn"
|
|
|
|
"github.com/stretchr/testify/assert"
|
|
"github.com/stretchr/testify/require"
|
|
)
|
|
|
|
var configEnvMu sync.Mutex
|
|
|
|
const (
|
|
gatewayRedisMasterAddrEnvVar = "GATEWAY_REDIS_MASTER_ADDR"
|
|
gatewayRedisPasswordEnvVar = "GATEWAY_REDIS_PASSWORD"
|
|
)
|
|
|
|
const (
|
|
defaultTestRedisMasterAddrValue = "127.0.0.1:6379"
|
|
defaultTestRedisPasswordValue = "secret"
|
|
defaultTestBackendHTTPURL = "http://127.0.0.1:8080"
|
|
defaultTestBackendGRPCPushURL = "127.0.0.1:8081"
|
|
defaultTestBackendClientID = "gw-test"
|
|
)
|
|
|
|
func TestLoadFromEnvAppliesBackendDefaults(t *testing.T) {
|
|
configEnvMu.Lock()
|
|
defer configEnvMu.Unlock()
|
|
|
|
resetEnv(t)
|
|
setBaseRequiredEnv(t)
|
|
|
|
cfg, err := LoadFromEnv()
|
|
require.NoError(t, err)
|
|
|
|
assert.Equal(t, defaultShutdownTimeout, cfg.ShutdownTimeout)
|
|
assert.Equal(t, defaultLogLevel, cfg.Logging.Level)
|
|
|
|
assert.Equal(t, defaultTestBackendHTTPURL, cfg.Backend.HTTPBaseURL)
|
|
assert.Equal(t, defaultTestBackendGRPCPushURL, cfg.Backend.GRPCPushURL)
|
|
assert.Equal(t, defaultTestBackendClientID, cfg.Backend.GatewayClientID)
|
|
assert.Equal(t, defaultBackendHTTPTimeout, cfg.Backend.HTTPTimeout)
|
|
assert.Equal(t, defaultBackendPushReconnectBaseBackoff, cfg.Backend.PushReconnectBaseBackoff)
|
|
assert.Equal(t, defaultBackendPushReconnectMaxBackoff, cfg.Backend.PushReconnectMaxBackoff)
|
|
|
|
expectedRedis := redisconn.DefaultConfig()
|
|
expectedRedis.MasterAddr = defaultTestRedisMasterAddrValue
|
|
expectedRedis.Password = defaultTestRedisPasswordValue
|
|
assert.Equal(t, expectedRedis, cfg.Redis)
|
|
|
|
assert.Equal(t, defaultReplayRedisKeyPrefix, cfg.ReplayRedis.KeyPrefix)
|
|
assert.Equal(t, defaultReplayRedisReserveTimeout, cfg.ReplayRedis.ReserveTimeout)
|
|
}
|
|
|
|
func TestLoadFromEnvBackendOverrides(t *testing.T) {
|
|
configEnvMu.Lock()
|
|
defer configEnvMu.Unlock()
|
|
|
|
resetEnv(t)
|
|
setBaseRequiredEnv(t)
|
|
|
|
t.Setenv(backendHTTPURLEnvVar, " http://backend.internal:9080/ ")
|
|
t.Setenv(backendGRPCPushURLEnvVar, "backend.internal:9081")
|
|
t.Setenv(backendGatewayClientIDEnvVar, "gw-prod-1")
|
|
t.Setenv(backendHTTPTimeoutEnvVar, "7s")
|
|
t.Setenv(backendPushReconnectBaseBackoffEnvVar, "750ms")
|
|
t.Setenv(backendPushReconnectMaxBackoffEnvVar, "60s")
|
|
|
|
cfg, err := LoadFromEnv()
|
|
require.NoError(t, err)
|
|
assert.Equal(t, "http://backend.internal:9080", cfg.Backend.HTTPBaseURL)
|
|
assert.Equal(t, "backend.internal:9081", cfg.Backend.GRPCPushURL)
|
|
assert.Equal(t, "gw-prod-1", cfg.Backend.GatewayClientID)
|
|
assert.Equal(t, 7*time.Second, cfg.Backend.HTTPTimeout)
|
|
assert.Equal(t, 750*time.Millisecond, cfg.Backend.PushReconnectBaseBackoff)
|
|
assert.Equal(t, time.Minute, cfg.Backend.PushReconnectMaxBackoff)
|
|
}
|
|
|
|
func TestLoadFromEnvRejectsMissingBackendValues(t *testing.T) {
|
|
cases := []struct {
|
|
name string
|
|
mutate func(t *testing.T)
|
|
wantErr string
|
|
}{
|
|
{
|
|
name: "http url missing",
|
|
mutate: func(t *testing.T) { os.Unsetenv(backendHTTPURLEnvVar) },
|
|
wantErr: backendHTTPURLEnvVar,
|
|
},
|
|
{
|
|
name: "grpc url missing",
|
|
mutate: func(t *testing.T) { os.Unsetenv(backendGRPCPushURLEnvVar) },
|
|
wantErr: backendGRPCPushURLEnvVar,
|
|
},
|
|
{
|
|
name: "gateway client id missing",
|
|
mutate: func(t *testing.T) { os.Unsetenv(backendGatewayClientIDEnvVar) },
|
|
wantErr: backendGatewayClientIDEnvVar,
|
|
},
|
|
{
|
|
name: "http url not absolute",
|
|
mutate: func(t *testing.T) { t.Setenv(backendHTTPURLEnvVar, "/relative") },
|
|
wantErr: backendHTTPURLEnvVar,
|
|
},
|
|
}
|
|
|
|
for _, tc := range cases {
|
|
t.Run(tc.name, func(t *testing.T) {
|
|
configEnvMu.Lock()
|
|
defer configEnvMu.Unlock()
|
|
|
|
resetEnv(t)
|
|
setBaseRequiredEnv(t)
|
|
tc.mutate(t)
|
|
|
|
_, err := LoadFromEnv()
|
|
require.Error(t, err)
|
|
assert.Contains(t, err.Error(), tc.wantErr)
|
|
})
|
|
}
|
|
}
|
|
|
|
func TestLoadFromEnvRejectsInvalidPushBackoff(t *testing.T) {
|
|
configEnvMu.Lock()
|
|
defer configEnvMu.Unlock()
|
|
|
|
resetEnv(t)
|
|
setBaseRequiredEnv(t)
|
|
t.Setenv(backendPushReconnectBaseBackoffEnvVar, "1s")
|
|
t.Setenv(backendPushReconnectMaxBackoffEnvVar, "500ms")
|
|
|
|
_, err := LoadFromEnv()
|
|
require.Error(t, err)
|
|
assert.Contains(t, err.Error(), backendPushReconnectMaxBackoffEnvVar)
|
|
}
|
|
|
|
func TestLoadFromEnvAppliesPublicAndAuthGRPCDefaults(t *testing.T) {
|
|
configEnvMu.Lock()
|
|
defer configEnvMu.Unlock()
|
|
|
|
resetEnv(t)
|
|
setBaseRequiredEnv(t)
|
|
|
|
cfg, err := LoadFromEnv()
|
|
require.NoError(t, err)
|
|
|
|
assert.Equal(t, defaultPublicHTTPAddr, cfg.PublicHTTP.Addr)
|
|
assert.Equal(t, defaultPublicHTTPReadHeaderTimeout, cfg.PublicHTTP.ReadHeaderTimeout)
|
|
assert.Equal(t, defaultPublicHTTPReadTimeout, cfg.PublicHTTP.ReadTimeout)
|
|
assert.Equal(t, defaultPublicHTTPIdleTimeout, cfg.PublicHTTP.IdleTimeout)
|
|
assert.Equal(t, defaultPublicAuthUpstreamTimeout, cfg.PublicHTTP.AuthUpstreamTimeout)
|
|
assert.Empty(t, cfg.PublicHTTP.CORSAllowedOrigins, "default disables CORS")
|
|
|
|
assert.Equal(t, defaultAuthenticatedGRPCAddr, cfg.AuthenticatedGRPC.Addr)
|
|
assert.Equal(t, defaultAuthenticatedGRPCConnectionTimeout, cfg.AuthenticatedGRPC.ConnectionTimeout)
|
|
assert.Equal(t, defaultAuthenticatedGRPCDownstreamTimeout, cfg.AuthenticatedGRPC.DownstreamTimeout)
|
|
assert.Equal(t, defaultAuthenticatedGRPCFreshnessWindow, cfg.AuthenticatedGRPC.FreshnessWindow)
|
|
assert.Equal(t, defaultPushHeartbeatInterval, cfg.AuthenticatedGRPC.PushHeartbeatInterval)
|
|
}
|
|
|
|
func TestLoadFromEnvParsesPushHeartbeatInterval(t *testing.T) {
|
|
configEnvMu.Lock()
|
|
defer configEnvMu.Unlock()
|
|
|
|
resetEnv(t)
|
|
setBaseRequiredEnv(t)
|
|
t.Setenv(pushHeartbeatIntervalEnvVar, "0s")
|
|
|
|
cfg, err := LoadFromEnv()
|
|
require.NoError(t, err)
|
|
assert.Equal(t, time.Duration(0), cfg.AuthenticatedGRPC.PushHeartbeatInterval, "0s explicitly disables the heartbeat")
|
|
|
|
t.Setenv(pushHeartbeatIntervalEnvVar, "25s")
|
|
cfg, err = LoadFromEnv()
|
|
require.NoError(t, err)
|
|
assert.Equal(t, 25*time.Second, cfg.AuthenticatedGRPC.PushHeartbeatInterval)
|
|
|
|
t.Setenv(pushHeartbeatIntervalEnvVar, "-1s")
|
|
_, err = LoadFromEnv()
|
|
require.Error(t, err)
|
|
assert.Contains(t, err.Error(), pushHeartbeatIntervalEnvVar)
|
|
}
|
|
|
|
func TestLoadFromEnvParsesCORSAllowedOrigins(t *testing.T) {
|
|
configEnvMu.Lock()
|
|
defer configEnvMu.Unlock()
|
|
|
|
resetEnv(t)
|
|
setBaseRequiredEnv(t)
|
|
t.Setenv(publicHTTPCORSAllowedOriginsEnvVar, "https://www.galaxy.lan, , https://staging.galaxy.lan")
|
|
|
|
cfg, err := LoadFromEnv()
|
|
require.NoError(t, err)
|
|
assert.Equal(t,
|
|
[]string{"https://www.galaxy.lan", "https://staging.galaxy.lan"},
|
|
cfg.PublicHTTP.CORSAllowedOrigins,
|
|
"comma-separated list is split, whitespace-trimmed, and empty segments dropped")
|
|
}
|
|
|
|
// resetEnv clears every env var the gateway config might read so that
|
|
// individual tests can build the exact environment they need without
|
|
// leakage from a previous test.
|
|
func resetEnv(t *testing.T) {
|
|
t.Helper()
|
|
|
|
for _, name := range []string{
|
|
shutdownTimeoutEnvVar,
|
|
logLevelEnvVar,
|
|
publicHTTPAddrEnvVar,
|
|
publicHTTPReadHeaderTimeoutEnvVar,
|
|
publicHTTPReadTimeoutEnvVar,
|
|
publicHTTPIdleTimeoutEnvVar,
|
|
publicAuthUpstreamTimeoutEnvVar,
|
|
publicHTTPCORSAllowedOriginsEnvVar,
|
|
backendHTTPURLEnvVar,
|
|
backendGRPCPushURLEnvVar,
|
|
backendGatewayClientIDEnvVar,
|
|
backendHTTPTimeoutEnvVar,
|
|
backendPushReconnectBaseBackoffEnvVar,
|
|
backendPushReconnectMaxBackoffEnvVar,
|
|
adminHTTPAddrEnvVar,
|
|
adminHTTPReadHeaderTimeoutEnvVar,
|
|
adminHTTPReadTimeoutEnvVar,
|
|
adminHTTPIdleTimeoutEnvVar,
|
|
authenticatedGRPCAddrEnvVar,
|
|
authenticatedGRPCConnectionTimeoutEnvVar,
|
|
authenticatedGRPCDownstreamTimeoutEnvVar,
|
|
authenticatedGRPCFreshnessWindowEnvVar,
|
|
pushHeartbeatIntervalEnvVar,
|
|
gatewayRedisMasterAddrEnvVar,
|
|
gatewayRedisPasswordEnvVar,
|
|
replayRedisKeyPrefixEnvVar,
|
|
replayRedisReserveTimeoutEnvVar,
|
|
responseSignerPrivateKeyPEMPathEnvVar,
|
|
} {
|
|
os.Unsetenv(name)
|
|
}
|
|
}
|
|
|
|
func setBaseRequiredEnv(t *testing.T) {
|
|
t.Helper()
|
|
|
|
t.Setenv(gatewayRedisMasterAddrEnvVar, defaultTestRedisMasterAddrValue)
|
|
t.Setenv(gatewayRedisPasswordEnvVar, defaultTestRedisPasswordValue)
|
|
t.Setenv(backendHTTPURLEnvVar, defaultTestBackendHTTPURL)
|
|
t.Setenv(backendGRPCPushURLEnvVar, defaultTestBackendGRPCPushURL)
|
|
t.Setenv(backendGatewayClientIDEnvVar, defaultTestBackendClientID)
|
|
t.Setenv(responseSignerPrivateKeyPEMPathEnvVar, writeTestResponseSignerPEMFile(t))
|
|
}
|
|
|
|
func writeTestResponseSignerPEMFile(t *testing.T) string {
|
|
t.Helper()
|
|
|
|
seed := sha256.Sum256([]byte("gateway-config-test-response-signer"))
|
|
privateKey := ed25519.NewKeyFromSeed(seed[:])
|
|
|
|
encodedPrivateKey, err := x509.MarshalPKCS8PrivateKey(privateKey)
|
|
require.NoError(t, err)
|
|
|
|
path := filepath.Join(t.TempDir(), "response-signer.pem")
|
|
require.NoError(t, os.WriteFile(path, pem.EncodeToMemory(&pem.Block{
|
|
Type: "PRIVATE KEY",
|
|
Bytes: encodedPrivateKey,
|
|
}), 0o600))
|
|
|
|
return path
|
|
}
|