Files
Ben Duncan 8ab96b29a1 fix: replicate every JetStream KV bucket across the cluster (#1117)
* fix: replicate every JetStream KV bucket across the cluster

A KV bucket created without a replica count gets one replica, and one
replica lives on one JetStream-chosen server recorded in no config.
Losing, restarting or partitioning that server makes the bucket return
"no responders available" from every node at once, so a single-node event
becomes a cluster-wide outage. The reconciler leader-lease buckets were
the worst case: acquiring a lease is a write, so a lease bucket without a
quorum cannot be acquired by anyone, and every reconciler sharing it stops
cluster-wide at exactly the moment a node has failed.

The cluster's node count is now declared once, in config.LoadConfig, and
read by the one helper every bucket creation goes through. Every process
that creates a bucket loads the cluster config first, so no service can
forget to pass it and no signature has to carry it. An undeclared size is
an error at creation rather than a silent fall back to one replica.

Existing buckets are raised rather than left behind: a service opening one
raises it, the daemon sweeps them all after formation, and
"spx admin kv replicas --repair" does it on demand. Raising reads the live
stream config and changes only the replica count, and never lowers one.

Removes the ~30 replicas/clusterSize parameters that only forwarded a
number, and JetStreamManager.UpdateReplicas, which set len(nodes) with no
clamp and would have failed outright on a cluster of six or more.

Proved by a repo-wide invariant test pinning bucket creation to the one
helper, a whole-cluster sweep in every multi-node nightly cell, and a new
kvquorum suite (cell 33) that takes away the node leading the most buckets
and requires the survivors to keep answering for all of them.

* chore: name cell 33 in the nightly dispatch input

* fix: a KV bucket that cannot be raised must still open

The first version of the replica fix treated a raise it could not place as a
hard error, and on a real three-node cluster that turned the fix into the
outage it prevents. A node whose config named a fourth server wanted four
replicas, NATS answered "no suitable peers for placement" for three, and the
daemon retried for its full five-minute budget without finishing bootstrap.
The bucket was present, quorate and serving on three nodes throughout. A
config naming a node before that node serves is the ordinary middle of
growing a cluster, so that is a state to tolerate rather than refuse.

The line is now drawn between the two cases instead of around both. Creating
a bucket below the cluster's replica count still fails, because that is the
defect and a fallback there reintroduces it where it is hardest to notice.
Failing to raise an existing bucket warns and hands the bucket back, and the
raise is retried on every open besides the daemon sweep and --repair. This
matters most in kvlease, which runs on every reconcile tick: refusing there
would stop every reconciler sharing a lease bucket that was working.

getOrCreateBucket now attaches before it creates. A create asking for more
replicas than the cluster can place is refused before the server looks at the
name, so with create-first "this bucket already exists" was an inference from
which error came back, and the wrong one comes back on a single-node server.

RaiseAllBucketReplicas attempts every bucket and joins the failures, and
--repair keeps going too. Stopping at the first would leave the rest of a
cluster short because one bucket could not be placed.

Also adds nats.ServerName/NodeFromServerName beside the template that renders
server_name, with a test tying them together. The kvquorum suite had mapped
stream peers to nodes by comparing them to node names, which they are not:
they are the node name behind a "spinifex-nats-" prefix, so nothing matched
and the node-loss test could not choose a victim.

* test: wait for the right NATS peer count after a node rejoins

/routez reports a node's routes to its peers, not its own membership, so a
fully formed three-node cluster is two peers per node. The kvquorum suite
waited for three and spent its whole five-minute rejoin budget on a cluster
that had already reformed. Every other caller passes len(Nodes)-1; the
helper's doc comment now says so.

The assertion the test exists for had already passed: node1 was leading 18 of
47 buckets when it was taken away, and node2 and node3 each still answered for
all 47.

* fix: replicate KV buckets across the largest odd count a cluster can place

A bucket was replicated across min(nodes, 5) servers, which stores a fourth
full copy on a four-node cluster and still survives exactly one loss — the
same tolerance R3 gives for three quarters of the write amplification. A Raft
majority needs an odd count, so the rule becomes the largest odd count the
cluster can place, capped at five: 1, 3, 3, 5, 5. It is one expression with two
entry points, so no caller can hold a stale copy of the policy and keep passing
after the policy changes.

Three retry loops could not tell a configuration fault from a cluster that had
yet to form. An undeclared cluster size never becomes declared by waiting, so
each of them spent its whole budget on one and then reported quorum as the
cause. They now stop on a permanent error and say what is actually wrong. This
is what the integration suite was hitting: it builds its gateway by hand and
never loads a cluster config, so the size was never declared and the IAM retry
sat in its sleep until the test timed out. Both ends are fixed.

The audit reported configuration alone. A bucket at three replicas with one
peer offline and one lagging is one failure from unavailable and read as ok.
BucketReport now carries the leader and a count of the peers JetStream itself
calls current and reachable, quorum is measured against that count, and
spx admin kv replicas exits 0, 1 or 2 so a deployment gate can tell a problem
--repair fixes from one needing a human.

Declaration is write-once: daemon.Start refuses a conflicting redeclaration
because it is about to create buckets, and it declares before JetStream init
rather than after it. config.LoadConfig warns instead, because it also reads
config files that are not this process's own cluster.

spx admin init --nodes 2 is refused — no replica count on two nodes both
survives a loss and accepts writes — and single-node replication is asserted in
the single e2e suite, so it runs in every nightly host permutation.

* fix: withhold the vLLM serving image until its artifact is published

The catalog offers ubuntu-26.04-vllm-serving-x86_64, and the qcow2 it names
404s on iso.mulgadc.com, so spx admin images import would fail for every
operator who tried it. Its GPU siblings are published; this one never was.

The entry moves to WithheldImages rather than being deleted, because the entry
is the definition — the URL, the checksum and the two tags resolveServingAMI
filters on. Publishing the artifact is then a move back, not a retype.
Bedrock already handles the image not being registered: resolveServingAMI
returns ErrServingAMINotFound, which is the honest answer while there is
nothing to launch.

The bedrock tag contract is still pinned, now against the map the definition
lives in, and it additionally asserts the entry is not offered for import. The
e2e reachability walk covers AvailableImages alone, so a withheld entry is out
of it by construction.

* fix: drop the cluster size from the call sites dev added

Two constructors on dev still pass a replica count that no longer exists as a
parameter: the cluster size is a declared process fact, so EKSServiceDeps has
no ClusterSize field and NewIAMServiceImpl takes three arguments. The merge is
textually clean and breaks the build, which is the shape of collision a
signature removal always has.

* docs: explain what an operator sees when a cluster server fails

The replication section said how many copies a bucket has and stopped there,
which leaves the question an operator actually has during an incident: one
server is gone, what still works. It now covers the majority side carrying on,
an isolated server refusing changes on purpose and why that is the behaviour
you want, guests continuing to run while anything that has to be recorded does
not, the risk of reading state from a server that is cut off, and a returning
server catching up by itself.

The surrounding prose reads from the operator's side rather than the
implementation's: what you see and what to do, not which consensus protocol is
underneath.
2026-09-28 12:34:55 -05:00

325 lines
14 KiB
Go

//go:build integration
// Package integration is the in-process integration test tier for spinifex's
// AWS-compatible control plane. Tests here start the REAL gateway router
// (gateway.SetupRoutes) against an embedded NATS JetStream instance shared
// across the whole package (see TestMain in main_test.go), with real IAM/STS
// services wired in for authentic SigV4 authz — the only thing faked is the
// daemon side of the wire: the NATS subjects a live spinifex daemon would
// answer (ec2.*, ebs.*, ...) are stubbed per-test via StubSubject.
//
// Nothing is provisioned: no tofu, no docker, no Spinifex daemons, no fixed
// ports. Every test connects into its own isolated NATS account on the
// shared server (see TestMain), so tests stay independent in state despite
// sharing the underlying process, and are safe to run concurrently on a
// shared build host.
//
// This package is gated behind the "integration" build tag (see the
// `test-integration` Makefile target) so it is never swept by the default
// `go test ./spinifex/...`, `make test-cover`, or `make test-race`.
package integration
import (
"encoding/json"
"net/http/httptest"
"testing"
"github.com/aws/aws-sdk-go/aws"
awscreds "github.com/aws/aws-sdk-go/aws/credentials"
awssession "github.com/aws/aws-sdk-go/aws/session"
"github.com/aws/aws-sdk-go/service/bedrock"
"github.com/aws/aws-sdk-go/service/ec2"
"github.com/aws/aws-sdk-go/service/ecr"
"github.com/aws/aws-sdk-go/service/iam"
"github.com/aws/aws-sdk-go/service/sts"
"github.com/mulgadc/spinifex/spinifex/gateway"
gateway_bedrock "github.com/mulgadc/spinifex/spinifex/gateway/bedrock"
gateway_ecr "github.com/mulgadc/spinifex/spinifex/gateway/ecr"
gateway_ecrauth "github.com/mulgadc/spinifex/spinifex/gateway/ecrauth"
handlers_ecr "github.com/mulgadc/spinifex/spinifex/handlers/ecr"
handlers_iam "github.com/mulgadc/spinifex/spinifex/handlers/iam"
handlers_sts "github.com/mulgadc/spinifex/spinifex/handlers/sts"
"github.com/mulgadc/spinifex/spinifex/objectstore"
"github.com/mulgadc/spinifex/spinifex/types"
"github.com/mulgadc/spinifex/spinifex/utils"
"github.com/nats-io/nats.go"
"github.com/stretchr/testify/require"
)
const (
// testRegion/testAZ pin the gateway's SigV4 credential scope and AZ.
// Every SDK client built via Gateway.EC2Client signs against testRegion,
// so requests verify against the same scope the gateway expects.
testRegion = "us-east-1"
testAZ = "us-east-1a"
// testAccessKeyID/testSecretAccessKey are the root credentials seeded
// into the real IAM service for every harness-started gateway. AKIA
// prefix routes SigV4 auth through the long-lived-key path (see
// spinifex/gateway/auth.go resolveLongLivedAKID).
testAccessKeyID = "AKIAINTEGRATIONTESTR"
testSecretAccessKey = "integration-test-root-secret"
// nodeDiscoverSubject is the fan-out GatewayConfig.DiscoverActiveNodes
// publishes to (spinifex/gateway/gateway.go). StartGateway stubs it with
// a single-node reply so any call path that triggers node discovery
// resolves immediately instead of riding out the fan-out's 500ms timeout.
nodeDiscoverSubject = "spinifex.nodes.discover"
// testECRAudience is the audience claim every harness-issued ECR JWT is
// minted for and verified against — arbitrary but stable within this tier,
// mirroring production's "ecr.<region>.<suffix>" shape without depending on
// GatewayConfig.InternalSuffix (unset here; the in-process harness talks to
// the gateway's httptest address directly rather than a registry hostname).
testECRAudience = "ecr.integration-test"
)
// Gateway is a running in-process instance of the real AWS gateway router,
// backed by the package's shared embedded NATS JetStream server and real
// IAM/STS services. Obtained via StartGateway; the httptest server and any
// StubSubject responders registered on it are torn down automatically via
// t.Cleanup. NATSConn is NOT closed at test end — see the TestMain doc
// comment in main_test.go for why — TestMain closes every test's connection
// together once the whole package's tests have finished.
type Gateway struct {
// Server hosts gateway.SetupRoutes() on an ephemeral httptest port.
Server *httptest.Server
// NATSConn is this test's isolated JetStream connection into the shared
// server (see TestMain), which the gateway, IAMService, and STSService
// all share. Tests use it directly with StubSubject to fake daemon-side
// responders.
NATSConn *nats.Conn
// AccountID is the account the seeded root credentials belong to
// (utils.GlobalAccountID).
AccountID string
// Config is the GatewayConfig SetupRoutes was built from, exposed so
// tests can inspect or further wire fields StartGateway did not set.
Config *gateway.GatewayConfig
}
// Option customises a Gateway's GatewayConfig before SetupRoutes builds the
// router. Applied after StartGateway's own defaults, so an Option can
// override any of them (e.g. a different Region/AZ, or a nil IAMService to
// exercise the pre-IAM-compatibility bypass paths).
type Option func(*gateway.GatewayConfig)
// StartGateway boots the real gateway router over the package's shared,
// embedded NATS JetStream server — connected via a fresh, isolated NATS
// account so this test's state cannot collide with any other's — and serves
// it from an httptest.Server on an ephemeral port. It wires real
// IAMService/STSService implementations (not a mock) and
// seeds a root access key via SeedBootstrap, so SigV4 requests authenticate
// and authorize exactly as they would against a live environment — root
// bypasses IAM policy evaluation, matching gateway.evaluatePrincipalPolicy.
//
// ExpectedNodes is pinned to 1: utils.Gather waits the FULL timeout on every
// call when ExpectedNodes is 0, which would make every stubbed NATS
// round-trip pathologically slow.
//
// Callers stub whichever NATS subjects their test needs via StubSubject;
// every other control-plane path (auth, routing, XML/JSON marshaling, IAM
// policy evaluation) runs for real.
func StartGateway(t *testing.T, opts ...Option) *Gateway {
return startGateway(t, suiteConformance, opts...)
}
func startGateway(t *testing.T, collector *conformanceCollector, opts ...Option) *Gateway {
t.Helper()
// Every test connects into its own dynamically provisioned NATS account
// on the one server this package's TestMain shares across the whole
// binary, rather than booting a private embedded server per test — see
// the TestMain doc comment in main_test.go for why, and how state stays
// isolated regardless.
nc, js, err := sharedNATSHarness.connectIsolated(accountNameFor(t))
require.NoError(t, err)
masterKey, err := handlers_iam.GenerateMasterKey()
require.NoError(t, err)
iamSvc, err := handlers_iam.NewIAMServiceWithRetry(t.Context(), nc, masterKey)
require.NoError(t, err)
stsSvc, err := handlers_sts.NewSTSServiceImpl(t.Context(), nc, iamSvc, masterKey)
require.NoError(t, err)
encryptedSecret, err := handlers_iam.EncryptSecret(testSecretAccessKey, masterKey)
require.NoError(t, err)
require.NoError(t, iamSvc.SeedBootstrap(&handlers_iam.BootstrapData{
AccessKeyID: testAccessKeyID,
EncryptedSecret: encryptedSecret,
AccountID: utils.GlobalAccountID,
}))
// ECR auth bridge signing key: reuses the IAM master key to encrypt the
// signing key at rest in the same embedded JetStream KV, matching
// production's awsgw-keys wiring (services/awsgw/awsgw.go).
signingKey, verifyKeys, err := gateway_ecrauth.LoadOrCreateSigningKey(t.Context(), js, masterKey)
require.NoError(t, err)
bedrockAccess := gateway_bedrock.NewModelAccessStore(js)
cfg := &gateway.GatewayConfig{
DisableLogging: true,
NATSConn: nc,
ExpectedNodes: 1,
Region: testRegion,
AZ: testAZ,
IAMService: iamSvc,
STSService: stsSvc,
// ECRRegistry serves the OCI Distribution v2 (/v2/*) data plane; its
// Meta store is a NATS client, so a test that pushes/pulls or manages
// repositories must additionally call StartECRDaemonLite to subscribe a
// real MetaServiceImpl or every ECR request will time out with no
// responder. Blob/manifest bytes are memory-backed: no predastore.
ECRRegistry: gateway_ecr.NewRegistry(objectstore.NewMemoryObjectStore(), handlers_ecr.NewNATSMetaStore(nc), utils.GlobalAccountID),
ECRTokenIssuer: gateway_ecrauth.NewIssuer(signingKey, testECRAudience),
ECRTokenVerifier: gateway_ecrauth.NewVerifier(verifyKeys, testECRAudience),
// Ochre model access is deny-by-default, so without a grant store every
// bedrock route refuses. Tests sign as the system account, which the
// store exempts from grants exactly as it does in production, so the
// catalog is visible here without provisioning grants per test; a test
// signing as a tenant account would have to grant first.
BedrockAccess: bedrockAccess,
BedrockAccessAdmin: bedrockAccess,
}
for _, opt := range opts {
opt(cfg)
}
handler := cfg.SetupRoutes()
if collector != nil {
handler = conformanceMiddleware(handler, collector)
}
srv := httptest.NewServer(handler)
t.Cleanup(srv.Close)
gw := &Gateway{
Server: srv,
NATSConn: nc,
AccountID: utils.GlobalAccountID,
Config: cfg,
}
nodeReply, err := json.Marshal(types.NodeDiscoverResponse{Node: "integration-test-node"})
require.NoError(t, err)
gw.StubSubject(t, nodeDiscoverSubject, nodeReply)
return gw
}
// EC2Client returns an aws-sdk-go v1 EC2 client pointed at the gateway's
// httptest server, signed with the seeded root credentials over plain HTTP
// (the httptest server is not TLS).
func (gw *Gateway) EC2Client(t *testing.T) *ec2.EC2 {
t.Helper()
return ec2.New(gw.session(t))
}
// IAMClient returns an aws-sdk-go v1 IAM client pointed at the gateway's
// httptest server, signed with the seeded root credentials.
func (gw *Gateway) IAMClient(t *testing.T) *iam.IAM {
t.Helper()
return iam.New(gw.session(t))
}
// STSClient returns an aws-sdk-go v1 STS client pointed at the gateway's
// httptest server, signed with the seeded root credentials.
func (gw *Gateway) STSClient(t *testing.T) *sts.STS {
t.Helper()
return sts.New(gw.session(t))
}
// ECRClient returns an aws-sdk-go v1 ECR client pointed at the gateway's
// httptest server, signed with the seeded root credentials. Repository and
// image operations dispatch to ECRRegistry.Meta over NATS, so a test needs
// StartECRDaemonLite running before calling anything beyond
// GetAuthorizationToken.
func (gw *Gateway) ECRClient(t *testing.T) *ecr.ECR {
t.Helper()
return ecr.New(gw.session(t))
}
// BedrockClient returns an aws-sdk-go v1 Bedrock (control-plane) client
// pointed at the gateway's httptest server, signed with the seeded root
// credentials. ListFoundationModels/GetFoundationModel are pure gateway-side
// catalog logic (gateway/bedrock/catalog.go) with no NATS/daemon hop, so no
// DaemonLite wiring is needed to exercise them.
func (gw *Gateway) BedrockClient(t *testing.T) *bedrock.Bedrock {
t.Helper()
return bedrock.New(gw.session(t))
}
// PrincipalClients bundles the SDK service handles an authz test needs to act
// as a principal other than the harness's seeded root user: an IAM user
// created within the test, or a role assumed via STS. Populated by
// ClientsWithCreds / ClientsWithSessionCreds.
type PrincipalClients struct {
EC2 *ec2.EC2
IAM *iam.IAM
STS *sts.STS
ECR *ecr.ECR
}
// ClientsWithCreds returns PrincipalClients signed with a static access
// key/secret and no session token — the long-lived-access-key path (an AKIA
// key from IAM CreateAccessKey), as opposed to the seeded-root session
// EC2Client/IAMClient/STSClient use.
func (gw *Gateway) ClientsWithCreds(t *testing.T, accessKeyID, secretAccessKey string) *PrincipalClients {
t.Helper()
return gw.principalClients(t, accessKeyID, secretAccessKey, "")
}
// ClientsWithSessionCreds returns PrincipalClients signed with STS temporary
// credentials — the assumed-role path (an ASIA key, secret, and session
// token from STS AssumeRole).
func (gw *Gateway) ClientsWithSessionCreds(t *testing.T, accessKeyID, secretAccessKey, sessionToken string) *PrincipalClients {
t.Helper()
return gw.principalClients(t, accessKeyID, secretAccessKey, sessionToken)
}
// principalClients builds the session backing both ClientsWithCreds and
// ClientsWithSessionCreds. Passing sessionToken == "" here — the
// ClientsWithCreds path — must not add an X-Amz-Security-Token header at
// all: aws-sdk-go's v4 signer only sets the header when SessionToken is
// non-empty, so a single code path correctly serves both the long-lived-key
// and assumed-role principals without a conditional.
func (gw *Gateway) principalClients(t *testing.T, accessKeyID, secretAccessKey, sessionToken string) *PrincipalClients {
t.Helper()
sess, err := awssession.NewSession(&aws.Config{
Region: aws.String(testRegion),
Endpoint: aws.String(gw.Server.URL),
Credentials: awscreds.NewStaticCredentials(accessKeyID, secretAccessKey, sessionToken),
DisableSSL: aws.Bool(true),
// Error-path tests assert on the first response; the default retry
// loop would mask a deterministic 4xx behind minutes of backoff.
MaxRetries: aws.Int(0),
})
require.NoError(t, err)
return &PrincipalClients{
EC2: ec2.New(sess),
IAM: iam.New(sess),
STS: sts.New(sess),
ECR: ecr.New(sess),
}
}
// session builds the shared aws-sdk-go v1 Session every per-service client
// accessor (EC2Client, ...) derives from, pointed at this Gateway's httptest
// server and signed with the seeded root credentials.
func (gw *Gateway) session(t *testing.T) *awssession.Session {
t.Helper()
sess, err := awssession.NewSession(&aws.Config{
Region: aws.String(testRegion),
Endpoint: aws.String(gw.Server.URL),
Credentials: awscreds.NewStaticCredentials(testAccessKeyID, testSecretAccessKey, ""),
DisableSSL: aws.Bool(true),
// Error-path tests assert on the first response; the default retry
// loop would mask a deterministic 4xx behind minutes of backoff.
MaxRetries: aws.Int(0),
})
require.NoError(t, err)
return sess
}