feat: append a -kafka image variant to every version the control plane hands a fleet node when the instance runs Kafka, so a joining worker pulls a build that can actually reach the bus instead of failing at boot (#449)

AGENTS.md

@@ -359,6 +359,7 @@ Auto-update:
 - the heartbeat reply carries `desired_version`; the node writes it to a file and a systemd timer (`warmbly-node-update`, installed by the join script) pulls and restarts. The process being replaced is never the process doing the replacing
 - an empty `desired_version` means "no opinion" and must never be read as "downgrade to nothing". A node that cannot be told what to run keeps running what it has
 - a per-node `pinned_version` overrides the fleet target, for canarying or holding a machine back
+- **the version names the build, not just the release.** The default images are CGO-free and carry no librdkafka, so a node running one cannot speak Kafka: it would take `EVENTBUS_PROVIDER=kafka` from its rendered env and fail at boot. `imageVariant` in `internal/app/fleetnode/service.go` appends `-kafka` to every version an instance on Kafka hands out, pins included, because the control plane is the only side that knows which bus it runs. The node needs no change for this: `join.sh` writes the resolved version to `WARMBLY_VERSION`, the node reports that back, and the updater compares against it, so the suffix stays consistent through join, heartbeat and self-update. `FLEET_IMAGE_VARIANT` overrides it, and set-and-empty disables it
 - **the backend is deliberately excluded.** It is what tells everyone else their version; a self-update that goes wrong leaves nothing to recover with
 
 The join script is `internal/api/handler/nodescript/join.sh`, embedded and served at `GET /join.sh` by the instance itself, so a self-hosted fleet never depends on a vendor host and always gets a script matching its backend. There is exactly one copy: do not add a mirror under `scripts/` or `site/public/`. All the POSIX-sh rules for published scripts apply to it (`sh -n`, `shellcheck -s sh`, everything in a function, `main "$@"` last).

docs/content/docs/development/configuration.mdx

@@ -290,6 +290,7 @@ On `filesystem`, a remote worker writes blobs to its own disk rather than a volu
 | Variable | What it does | Default | Restart needed |
 |---|---|---|---|
 | `EVENTBUS_PROVIDER` | `nats` or `kafka`. Kafka needs the `-kafka` images, or a build with `GO_TAGS=kafka` | `nats` under compose, `kafka` for a bare binary | yes |
+| `FLEET_IMAGE_VARIANT` | Tag suffix appended to the version handed to fleet nodes, so they pull a build that can talk to this instance's event bus. Defaults to `-kafka` when `EVENTBUS_PROVIDER=kafka` and nothing otherwise; set it empty to turn it off | derived from `EVENTBUS_PROVIDER` | no |
 | `NATS_URL` | JetStream address. Credentials in the URL are honored by every service, including the Rust tracking publisher: `nats://user:pass@host:4222` for a user, `nats://token@host:4222` for a token, `tls://` for TLS | `nats://nats:4222` | yes |
 | `NATS_CREDS` | Path to a NATS credentials file (user JWT plus nkey seed), for a bus that authenticates with JWT rather than a token. Synadia Cloud and any nsc-managed account issue one | unset | yes |
 | `NATS_CREDS_B64` | The same file, base64 encoded, as a single line. This is the form the fleet uses: a node receives environment variables rather than files, and the env file docker reads cannot express a multi-line value. Takes precedence over `NATS_CREDS` | unset | yes |

docs/content/docs/development/deployment-guide.mdx

@@ -780,6 +780,8 @@ warmblyctl fleet pin <node> v1.4.1  # hold or canary one machine
 
 Setting a tag also pins the channel, so a release landing later does not silently undo a deliberate rollback. `fleet channel stable` resumes following releases.
 
+The version a node is told to run also names which build of the image it runs. An instance on `EVENTBUS_PROVIDER=kafka` hands its nodes `v1.4.2-kafka`, because the default images are CGO-free and carry no librdkafka: a node running one would take `EVENTBUS_PROVIDER=kafka` from its config and fail at boot. The suffix is added to whatever the version resolved to, including a pin, so `fleet pin <node> v1.4.1` still reaches the machine as something it can run. Set `FLEET_IMAGE_VARIANT` on the backend to change or disable it if you publish your own images under a different convention; setting it empty turns it off.
+
 The backend is deliberately excluded: it is the thing that tells every node what version to be, so it is upgraded the same way as the rest of your infrastructure. Upgrade the backend, and the fleet follows.
 
 ### Watching the fleet

internal/app/fleetnode/service.go

@@ -12,6 +12,7 @@ import (
 	"crypto/subtle"
 	"encoding/base64"
 	"errors"
+	"os"
 	"strings"
 
 	"github.com/google/uuid"
@@ -37,6 +38,11 @@ type Service struct {
 	nodes    repository.FleetNodeRepository
 	workers  repository.WorkerRepository
 	settings repository.FleetSettingsRepository
+
+	// variant is appended to every version this service hands a node, because
+	// a version names an image and some builds of an image are not
+	// interchangeable. See imageVariant.
+	variant string
 }
 
 func New(
@@ -44,7 +50,39 @@ func New(
 	workers repository.WorkerRepository,
 	settings repository.FleetSettingsRepository,
 ) *Service {
-	return &Service{nodes: nodes, workers: workers, settings: settings}
+	return &Service{nodes: nodes, workers: workers, settings: settings, variant: imageVariant()}
+}
+
+// imageVariant is the tag suffix a node must add to reach an image that can
+// talk to this instance's event bus.
+//
+// The default images are CGO-free and carry no librdkafka, so a node running
+// one cannot speak Kafka at all: it would take EVENTBUS_PROVIDER=kafka from
+// its rendered env and fail at boot. The Kafka builds are published under the
+// same image name with a "-kafka" tag suffix, so the fix is to name that tag,
+// and the control plane is the only side that knows which bus it runs.
+//
+// FLEET_IMAGE_VARIANT overrides it for anyone publishing their own images
+// under a different convention. Set and empty means "no suffix", which is how
+// an operator whose own Kafka build is tagged plainly opts out.
+func imageVariant() string {
+	if v, ok := os.LookupEnv("FLEET_IMAGE_VARIANT"); ok {
+		return v
+	}
+	if strings.EqualFold(os.Getenv("EVENTBUS_PROVIDER"), "kafka") {
+		return "-kafka"
+	}
+	return ""
+}
+
+// withVariant appends the image variant to a resolved version. An empty
+// version stays empty: "no opinion" must never become a bare "-kafka", which
+// the node would dutifully try to pull.
+func (s *Service) withVariant(version string) string {
+	if version == "" || s.variant == "" || strings.HasSuffix(version, s.variant) {
+		return version
+	}
+	return version + s.variant
 }
 
 // IssueJoinToken mints a new instance join token, stores only its hash, and
@@ -135,13 +173,13 @@ func (s *Service) Heartbeat(ctx context.Context, beat models.NodeHeartbeat) (*mo
 // hiccup rolling the whole fleet.
 func (s *Service) desiredVersion(ctx context.Context, nodeID uuid.UUID) string {
 	if node, err := s.nodes.Get(ctx, nodeID); err == nil && node != nil && node.PinnedVersion != "" {
-		return node.PinnedVersion
+		return s.withVariant(node.PinnedVersion)
 	}
 	state, err := s.settings.GetRelease(ctx)
 	if err != nil {
 		return ""
 	}
-	return state.DesiredVersion()
+	return s.withVariant(state.DesiredVersion())
 }
 
 // List returns the fleet, with each node's resolved target attached so a
@@ -155,10 +193,10 @@ func (s *Service) List(ctx context.Context, role models.NodeRole) ([]models.Flee
 	if err != nil {
 		return nil, err
 	}
-	fleetTarget := state.DesiredVersion()
+	fleetTarget := s.withVariant(state.DesiredVersion())
 	for i := range nodes {
 		if nodes[i].PinnedVersion != "" {
-			nodes[i].DesiredVersion = nodes[i].PinnedVersion
+			nodes[i].DesiredVersion = s.withVariant(nodes[i].PinnedVersion)
 			continue
 		}
 		nodes[i].DesiredVersion = fleetTarget
@@ -173,13 +211,13 @@ func (s *Service) Get(ctx context.Context, id uuid.UUID) (*models.FleetNode, err
 		return nil, err
 	}
 	if node.PinnedVersion != "" {
-		node.DesiredVersion = node.PinnedVersion
+		node.DesiredVersion = s.withVariant(node.PinnedVersion)
 		return node, nil
 	}
 	state, err := s.settings.GetRelease(ctx)
 	if err != nil {
 		return nil, err
 	}
-	node.DesiredVersion = state.DesiredVersion()
+	node.DesiredVersion = s.withVariant(state.DesiredVersion())
 	return node, nil
 }

internal/app/fleetnode/service_test.go

@@ -0,0 +1,151 @@
+package fleetnode
+
+import (
+	"context"
+	"os"
+	"testing"
+
+	"github.com/google/uuid"
+	"github.com/warmbly/warmbly/internal/models"
+	"github.com/warmbly/warmbly/internal/repository"
+)
+
+// os_Unsetenv clears a variable for the duration of the test; t.Setenv already
+// restores whatever was there.
+func os_Unsetenv(t *testing.T, key string) {
+	t.Helper()
+	t.Setenv(key, "")
+	if err := os.Unsetenv(key); err != nil {
+		t.Fatal(err)
+	}
+}
+
+// Only the two reads the version resolution makes are implemented; anything
+// else would panic, which is the point.
+type stubNodes struct {
+	repository.FleetNodeRepository
+	node *models.FleetNode
+}
+
+func (s stubNodes) Get(context.Context, uuid.UUID) (*models.FleetNode, error) {
+	return s.node, nil
+}
+
+func (s stubNodes) List(context.Context, models.NodeRole) ([]models.FleetNode, error) {
+	return []models.FleetNode{*s.node}, nil
+}
+
+type stubSettings struct {
+	repository.FleetSettingsRepository
+	release string
+}
+
+func (s stubSettings) GetRelease(context.Context) (*models.FleetReleaseState, error) {
+	return &models.FleetReleaseState{Tag: s.release}, nil
+}
+
+func TestImageVariant(t *testing.T) {
+	cases := []struct {
+		name     string
+		bus      string
+		override string
+		hasOver  bool
+		want     string
+	}{
+		{name: "nats gets no suffix", bus: "nats", want: ""},
+		{name: "unset bus gets no suffix", want: ""},
+		{name: "kafka gets the kafka suffix", bus: "kafka", want: "-kafka"},
+		{name: "kafka is matched case insensitively", bus: "KAFKA", want: "-kafka"},
+		{name: "an explicit override wins", bus: "kafka", override: "-confluent", hasOver: true, want: "-confluent"},
+		// Set and empty is how an operator whose own Kafka build is tagged
+		// plainly opts out; it must not fall through to the default.
+		{name: "an explicit empty override disables it", bus: "kafka", override: "", hasOver: true, want: ""},
+	}
+	for _, tc := range cases {
+		t.Run(tc.name, func(t *testing.T) {
+			t.Setenv("EVENTBUS_PROVIDER", tc.bus)
+			if tc.hasOver {
+				t.Setenv("FLEET_IMAGE_VARIANT", tc.override)
+			} else {
+				os_Unsetenv(t, "FLEET_IMAGE_VARIANT")
+			}
+			if got := imageVariant(); got != tc.want {
+				t.Fatalf("imageVariant() = %q, want %q", got, tc.want)
+			}
+		})
+	}
+}
+
+func TestWithVariant(t *testing.T) {
+	s := &Service{variant: "-kafka"}
+	cases := []struct{ in, want string }{
+		{"v0.4.5", "v0.4.5-kafka"},
+		// "No opinion" must stay no opinion. A bare "-kafka" is a tag the node
+		// would dutifully try to pull.
+		{"", ""},
+		// Idempotent, so a pin an operator already wrote with the suffix does
+		// not become v0.4.5-kafka-kafka.
+		{"v0.4.5-kafka", "v0.4.5-kafka"},
+	}
+	for _, tc := range cases {
+		if got := s.withVariant(tc.in); got != tc.want {
+			t.Errorf("withVariant(%q) = %q, want %q", tc.in, got, tc.want)
+		}
+	}
+	plain := &Service{variant: ""}
+	if got := plain.withVariant("v0.4.5"); got != "v0.4.5" {
+		t.Errorf("no variant should pass the version through, got %q", got)
+	}
+}
+
+// A pin is the sharp edge: an operator canarying a node types "v0.4.4", and
+// that has to reach the machine as the image the machine can actually run.
+func TestDesiredVersionAppliesVariantToPinsAndFleetTarget(t *testing.T) {
+	id := uuid.New()
+	settings := stubSettings{release: "v0.4.5"}
+
+	pinned := &Service{
+		nodes:    stubNodes{node: &models.FleetNode{ID: id, PinnedVersion: "v0.4.4"}},
+		settings: settings,
+		variant:  "-kafka",
+	}
+	if got := pinned.desiredVersion(context.Background(), id); got != "v0.4.4-kafka" {
+		t.Errorf("pinned node: got %q, want v0.4.4-kafka", got)
+	}
+
+	unpinned := &Service{
+		nodes:    stubNodes{node: &models.FleetNode{ID: id}},
+		settings: settings,
+		variant:  "-kafka",
+	}
+	if got := unpinned.desiredVersion(context.Background(), id); got != "v0.4.5-kafka" {
+		t.Errorf("unpinned node: got %q, want v0.4.5-kafka", got)
+	}
+}
+
+// List and Get feed the admin panel's "is this machine behind" column. A node
+// reports the WARMBLY_VERSION the join script wrote, which already carries the
+// suffix, so the target it is compared against has to carry it too or every
+// Kafka node reads as permanently out of date.
+func TestListAndGetAttachTheVariant(t *testing.T) {
+	id := uuid.New()
+	s := &Service{
+		nodes:    stubNodes{node: &models.FleetNode{ID: id}},
+		settings: stubSettings{release: "v0.4.5"},
+		variant:  "-kafka",
+	}
+	nodes, err := s.List(context.Background(), models.NodeRoleWorker)
+	if err != nil {
+		t.Fatal(err)
+	}
+	if nodes[0].DesiredVersion != "v0.4.5-kafka" {
+		t.Errorf("List: got %q, want v0.4.5-kafka", nodes[0].DesiredVersion)
+	}
+	node, err := s.Get(context.Background(), id)
+	if err != nil {
+		t.Fatal(err)
+	}
+	if node.DesiredVersion != "v0.4.5-kafka" {
+		t.Errorf("Get: got %q, want v0.4.5-kafka", node.DesiredVersion)
+	}
+}