From 6504d9958a6fb2a7220659f911e8927506a56b15 Mon Sep 17 00:00:00 2001 From: Matthew Meszaros Date: Sat, 12 Sep 2026 06:02:49 +0200 Subject: [PATCH 1/2] feat: create Kafka topics from the bus that uses them, because a worker's command topic is named after the node id issued at join time so the set is not knowable in advance, and the broker's auto-creation is off by default on Confluent Cloud and not configurable below Standard, which left a worker subscribed to a topic that did not exist receiving nothing and reporting no error, and compile the tagged Kafka build in CI so a backend nothing else builds cannot rot unnoticed --- .github/workflows/ci.yml | 6 + Makefile | 6 +- .../docs/development/configuration.mdx | 5 + internal/infrastructure/eventbus/kafka.go | 26 ++++ .../infrastructure/eventbus/kafka_topics.go | 116 ++++++++++++++++++ .../eventbus/kafka_topics_test.go | 49 ++++++++ 6 files changed, 207 insertions(+), 1 deletion(-) create mode 100644 internal/infrastructure/eventbus/kafka_topics.go create mode 100644 internal/infrastructure/eventbus/kafka_topics_test.go diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index dc65a8bd..946f7828 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -136,6 +136,12 @@ jobs: - name: Verify protobuf files are up to date run: make check-proto + # The Kafka backend sits behind a build tag, so the default build never + # compiles it and it can rot unnoticed. Vet rather than test: it needs + # CGO and a broker, but a compile error is the failure that matters. + - name: Verify the Kafka build still compiles + run: go vet -tags kafka ./... + - name: Run tests with coverage run: | go test -race -coverprofile=coverage.out -covermode=atomic ./... diff --git a/Makefile b/Makefile index 46f1934c..1c0e2f35 100644 --- a/Makefile +++ b/Makefile @@ -37,7 +37,7 @@ PROTOC_GEN_GO_GRPC_VERSION ?= v1.6.1 PROTO_DIR := internal/tasks/proto PROTO_GEN_FILES := $(PROTO_DIR)/tasks.pb.go -.PHONY: poollink-dev poollink-dev-down poollink-dev-reset setup-tools fmt lint check-migrations join-check split-cloud-check pages-check proto check-proto \ +.PHONY: poollink-dev poollink-dev-down poollink-dev-reset setup-tools fmt lint check-migrations join-check split-cloud-check pages-check kafka-check proto check-proto \ up upgrade claim doctor cli seed-demo seed seed-plan sandbox sandbox-seed sandbox-simulate reset logs status stop down test-seed \ restart restart-go restart-all infra infra-down app app-down app-logs \ backend forms forms-web consumer worker run dev tracking realtime web \ @@ -99,6 +99,10 @@ check-migrations: split-cloud-check: @./scripts/check-split-cloud.sh +# The Kafka backend is behind a build tag, so nothing else compiles it. +kafka-check: + go vet -tags kafka ./... + # web and admin on a static host. They read their configuration from a # config.js the container entrypoint renders at start, and a static host has no # container start, so the same script renders it at build time. This runs both diff --git a/docs/content/docs/development/configuration.mdx b/docs/content/docs/development/configuration.mdx index 7fbe3ae2..cca83b6f 100644 --- a/docs/content/docs/development/configuration.mdx +++ b/docs/content/docs/development/configuration.mdx @@ -303,6 +303,11 @@ On `filesystem`, a remote worker writes blobs to its own disk rather than a volu | `PUBSUB_ENABLED` | `false` uses the Redis bridge for realtime fanout, `true` uses Google Pub/Sub | `false` | yes | | `GCP_PROJECT_ID` | Project when `PUBSUB_ENABLED=true` | unset | yes | + +A worker's command topic is named after the node id issued when it joined, so the set of topics is not knowable in advance. The Kafka backend creates what it uses, once per topic per process, rather than depending on the broker's `auto.create.topics.enable` — which is off by default on Confluent Cloud and not configurable below Standard. The credential therefore needs `CREATE` on topics as well as read and write. + + + `CODEC_PROVIDER=json` is required wherever workers run: the worker command and result envelopes carry untyped bodies Avro cannot serialize, so any other value makes every worker command fail to encode. `PUBSUB_ENABLED` must agree across backend, consumer and realtime. diff --git a/internal/infrastructure/eventbus/kafka.go b/internal/infrastructure/eventbus/kafka.go index 479dc2a2..95aa7dc9 100644 --- a/internal/infrastructure/eventbus/kafka.go +++ b/internal/infrastructure/eventbus/kafka.go @@ -34,6 +34,10 @@ type KafkaBus struct { mu sync.Mutex consumers []*kafka.Consumer closed bool + + // Kafka topics must exist before use, and a worker's command topic is + // named after a node id issued at join time. See kafka_topics.go. + topics topicEnsurer } // NewKafka constructs a KafkaBus and opens the shared producer connection. @@ -53,6 +57,7 @@ func NewKafka(cfg KafkaConfig) (*KafkaBus, error) { producer: prod, bootstrap: cfg.Bootstrap, sasl: cfg.SASL, + topics: topicEnsurer{known: map[string]struct{}{}}, }, nil } @@ -67,6 +72,7 @@ func NewKafkaFromProducer(p *kafka.Producer, cfg KafkaConfig) *KafkaBus { producer: p, bootstrap: cfg.Bootstrap, sasl: cfg.SASL, + topics: topicEnsurer{known: map[string]struct{}{}}, } } @@ -86,6 +92,11 @@ func (b *KafkaBus) Publish(ctx context.Context, topic, key string, payload []byt if closed { return errors.New("eventbus kafka: bus closed") } + // A worker's command topic is named after its node id, so the first + // publish to it is the first time anything knows the name. + if err := b.ensureTopics(ctx, topic); err != nil { + return err + } return b.producer.Produce(topic, []byte(key), payload) } @@ -104,6 +115,12 @@ func (b *KafkaBus) Subscribe(ctx context.Context, topics []string, group string, return errors.New("eventbus kafka: handler required") } + // Subscribing to a topic that does not exist yet returns no messages and + // no error, so a worker would sit silent rather than fail. + if err := b.ensureTopics(ctx, topics...); err != nil { + return err + } + cc := kafka.NewConsumer(b.bootstrap) if b.sasl != nil { cc.WithSASL(b.sasl) @@ -152,6 +169,15 @@ func (b *KafkaBus) Subscribe(ctx context.Context, topics []string, group string, // Close flushes the producer and closes every consumer that was opened via // Subscribe. func (b *KafkaBus) Close() error { + // The admin client is a separate connection; Close is the only place that + // knows it was ever opened. + b.topics.mu.Lock() + if b.topics.admin != nil { + b.topics.admin.Close() + b.topics.admin = nil + } + b.topics.mu.Unlock() + b.mu.Lock() if b.closed { b.mu.Unlock() diff --git a/internal/infrastructure/eventbus/kafka_topics.go b/internal/infrastructure/eventbus/kafka_topics.go new file mode 100644 index 00000000..c7c5b885 --- /dev/null +++ b/internal/infrastructure/eventbus/kafka_topics.go @@ -0,0 +1,116 @@ +//go:build kafka + +package eventbus + +import ( + "context" + "fmt" + "strings" + "sync" + "time" + + ckf "github.com/confluentinc/confluent-kafka-go/v2/kafka" + "github.com/rs/zerolog/log" +) + +// Kafka has no equivalent of a NATS subject: a topic must exist before anything +// can be produced to it. Most of Warmbly's topics are fixed, but a worker's +// command topic is named after the node id it was given at join time, so the +// set is not knowable in advance and cannot be created by hand. +// +// Relying on the broker's auto.create.topics.enable is not enough: it is off by +// default on Confluent Cloud and not configurable below Standard, and a worker +// whose topic was never created receives no commands and reports no error. +// +// So the bus creates what it uses, once per topic per process. + +const ( + topicPartitions = 3 + topicReplicationFactor = 3 + topicAdminTimeout = 20 * time.Second +) + +type topicEnsurer struct { + mu sync.Mutex + known map[string]struct{} + admin *ckf.AdminClient +} + +// ensureTopics creates any topic in names that this process has not already +// created. Creating one that exists is not an error: the broker answers +// TOPIC_ALREADY_EXISTS and that is the common case after the first call. +func (b *KafkaBus) ensureTopics(ctx context.Context, names ...string) error { + b.topics.mu.Lock() + defer b.topics.mu.Unlock() + + var missing []ckf.TopicSpecification + for _, n := range names { + if n == "" { + continue + } + if _, seen := b.topics.known[n]; seen { + continue + } + missing = append(missing, ckf.TopicSpecification{ + Topic: n, + NumPartitions: topicPartitions, + ReplicationFactor: topicReplicationFactor, + }) + } + if len(missing) == 0 { + return nil + } + + admin, err := b.adminClient() + if err != nil { + return err + } + + cctx, cancel := context.WithTimeout(ctx, topicAdminTimeout) + defer cancel() + results, err := admin.CreateTopics(cctx, missing) + if err != nil { + return fmt.Errorf("eventbus kafka: create topics: %w", err) + } + for _, r := range results { + switch r.Error.Code() { + case ckf.ErrNoError: + log.Info().Str("topic", r.Topic).Msg("eventbus kafka: topic created") + case ckf.ErrTopicAlreadyExists: + // The steady state on every process after the first. + default: + // A replication factor the cluster cannot satisfy is the usual + // cause on a single-broker development cluster, and the bare + // error does not say so. + hint := "" + if strings.Contains(strings.ToLower(r.Error.String()), "replication") { + hint = " (the cluster has fewer brokers than the replication factor; this is expected on a single-broker development cluster)" + } + return fmt.Errorf("eventbus kafka: create topic %q: %s%s", r.Topic, r.Error.String(), hint) + } + b.topics.known[r.Topic] = struct{}{} + } + return nil +} + +// adminClient opens the admin connection lazily, so a deployment that never +// needs to create a topic never opens one. +func (b *KafkaBus) adminClient() (*ckf.AdminClient, error) { + if b.topics.admin != nil { + return b.topics.admin, nil + } + conf := &ckf.ConfigMap{"bootstrap.servers": b.bootstrap} + if b.sasl != nil { + // The same renderer the producer and consumer use, so the admin + // connection cannot authenticate differently from the traffic. + for k, v := range b.sasl.Generate() { + _ = conf.SetKey(k, v) + } + } + admin, err := ckf.NewAdminClient(conf) + if err != nil { + return nil, fmt.Errorf("eventbus kafka: admin client: %w", err) + } + b.topics.admin = admin + return admin, nil +} diff --git a/internal/infrastructure/eventbus/kafka_topics_test.go b/internal/infrastructure/eventbus/kafka_topics_test.go new file mode 100644 index 00000000..1a0c630b --- /dev/null +++ b/internal/infrastructure/eventbus/kafka_topics_test.go @@ -0,0 +1,49 @@ +//go:build kafka + +package eventbus + +import ( + "context" + "testing" +) + +// A topic already created by this process must not reach the broker again: +// ensureTopics runs on the publish path, and an admin round trip per message +// would put a network call in front of every send. +func TestEnsureTopicsSkipsKnown(t *testing.T) { + b := &KafkaBus{ + bootstrap: "localhost:0", + topics: topicEnsurer{known: map[string]struct{}{"jobs.worker-events": {}}}, + } + // No admin client is opened, so a broker call would fail rather than hang. + if err := b.ensureTopics(context.Background(), "jobs.worker-events"); err != nil { + t.Fatalf("a known topic tried to reach the broker: %v", err) + } + if b.topics.admin != nil { + t.Error("an admin connection was opened for a topic already known") + } +} + +// Empty names come from callers that build a topic from an id that is not set +// yet; they must not become a create request for "". +func TestEnsureTopicsIgnoresEmpty(t *testing.T) { + b := &KafkaBus{ + bootstrap: "localhost:0", + topics: topicEnsurer{known: map[string]struct{}{}}, + } + if err := b.ensureTopics(context.Background(), "", ""); err != nil { + t.Fatalf("empty topic names produced a request: %v", err) + } + if b.topics.admin != nil { + t.Error("an admin connection was opened for empty topic names") + } +} + +// Both constructors must initialise the map, or the first ensure panics on a +// nil-map write rather than failing to publish. +func TestBothConstructorsInitialiseTopicMap(t *testing.T) { + b := NewKafkaFromProducer(nil, KafkaConfig{Bootstrap: "localhost:0"}) + if b.topics.known == nil { + t.Fatal("NewKafkaFromProducer left the topic map nil") + } +} From e3092f2322da0758571a62c9d9304241a1e890df Mon Sep 17 00:00:00 2001 From: Matthew Meszaros Date: Sat, 12 Sep 2026 06:54:34 +0200 Subject: [PATCH 2/2] feat: address the review by releasing the topic lock before the broker call so one slow creation cannot stall every publish, subscribe and close for the full admin timeout, refusing to open an admin connection once the bus is closed so a publish racing past the closed check cannot resurrect a client nothing will shut, building rather than vetting the tagged Kafka backend in CI because vet does not link and a CGO backend fails at link time, and correcting the Confluent tier wording to say auto topic creation is configurable only on Dedicated --- .github/workflows/ci.yml | 8 +- Makefile | 3 +- .../docs/development/configuration.mdx | 2 +- internal/infrastructure/eventbus/kafka.go | 5 +- .../infrastructure/eventbus/kafka_topics.go | 78 +++++++++++++------ .../eventbus/kafka_topics_test.go | 45 +++++++++++ 6 files changed, 110 insertions(+), 31 deletions(-) diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 946f7828..4e623026 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -137,10 +137,10 @@ jobs: run: make check-proto # The Kafka backend sits behind a build tag, so the default build never - # compiles it and it can rot unnoticed. Vet rather than test: it needs - # CGO and a broker, but a compile error is the failure that matters. - - name: Verify the Kafka build still compiles - run: go vet -tags kafka ./... + # compiles it and it can rot unnoticed. Build rather than vet: vet does + # not link, and a CGO backend's characteristic failure is a link error. + - name: Verify the Kafka build still compiles and links + run: go build -tags kafka ./... - name: Run tests with coverage run: | diff --git a/Makefile b/Makefile index 1c0e2f35..8e4f0ff8 100644 --- a/Makefile +++ b/Makefile @@ -100,8 +100,9 @@ split-cloud-check: @./scripts/check-split-cloud.sh # The Kafka backend is behind a build tag, so nothing else compiles it. +# Build, not vet: vet does not link, and this backend links librdkafka. kafka-check: - go vet -tags kafka ./... + go build -tags kafka ./... # web and admin on a static host. They read their configuration from a # config.js the container entrypoint renders at start, and a static host has no diff --git a/docs/content/docs/development/configuration.mdx b/docs/content/docs/development/configuration.mdx index cca83b6f..b90c7191 100644 --- a/docs/content/docs/development/configuration.mdx +++ b/docs/content/docs/development/configuration.mdx @@ -304,7 +304,7 @@ On `filesystem`, a remote worker writes blobs to its own disk rather than a volu | `GCP_PROJECT_ID` | Project when `PUBSUB_ENABLED=true` | unset | yes | -A worker's command topic is named after the node id issued when it joined, so the set of topics is not knowable in advance. The Kafka backend creates what it uses, once per topic per process, rather than depending on the broker's `auto.create.topics.enable` — which is off by default on Confluent Cloud and not configurable below Standard. The credential therefore needs `CREATE` on topics as well as read and write. +A worker's command topic is named after the node id issued when it joined, so the set of topics is not knowable in advance. The Kafka backend creates what it uses, once per topic per process, rather than depending on the broker's `auto.create.topics.enable` — which is off by default on Confluent Cloud and configurable only on Dedicated clusters, not on Basic, Standard, Enterprise or Freight. The credential therefore needs `CREATE` on topics as well as read and write. diff --git a/internal/infrastructure/eventbus/kafka.go b/internal/infrastructure/eventbus/kafka.go index 95aa7dc9..b1c0b1cd 100644 --- a/internal/infrastructure/eventbus/kafka.go +++ b/internal/infrastructure/eventbus/kafka.go @@ -169,9 +169,10 @@ func (b *KafkaBus) Subscribe(ctx context.Context, topics []string, group string, // Close flushes the producer and closes every consumer that was opened via // Subscribe. func (b *KafkaBus) Close() error { - // The admin client is a separate connection; Close is the only place that - // knows it was ever opened. + // Marked closed before the client is cleared, so a Publish racing past the + // closed check cannot open a replacement that outlives shutdown. b.topics.mu.Lock() + b.topics.closed = true if b.topics.admin != nil { b.topics.admin.Close() b.topics.admin = nil diff --git a/internal/infrastructure/eventbus/kafka_topics.go b/internal/infrastructure/eventbus/kafka_topics.go index c7c5b885..cbc826e1 100644 --- a/internal/infrastructure/eventbus/kafka_topics.go +++ b/internal/infrastructure/eventbus/kafka_topics.go @@ -4,6 +4,7 @@ package eventbus import ( "context" + "errors" "fmt" "strings" "sync" @@ -31,34 +32,22 @@ const ( ) type topicEnsurer struct { - mu sync.Mutex - known map[string]struct{} - admin *ckf.AdminClient + // mu guards known, admin and closed. It is never held across the broker + // call: CreateTopics can wait the full admin timeout, and holding it there + // would stall Publish, Subscribe and Close for every other topic. + mu sync.Mutex + known map[string]struct{} + admin *ckf.AdminClient + closed bool } // ensureTopics creates any topic in names that this process has not already // created. Creating one that exists is not an error: the broker answers // TOPIC_ALREADY_EXISTS and that is the common case after the first call. func (b *KafkaBus) ensureTopics(ctx context.Context, names ...string) error { - b.topics.mu.Lock() - defer b.topics.mu.Unlock() - - var missing []ckf.TopicSpecification - for _, n := range names { - if n == "" { - continue - } - if _, seen := b.topics.known[n]; seen { - continue - } - missing = append(missing, ckf.TopicSpecification{ - Topic: n, - NumPartitions: topicPartitions, - ReplicationFactor: topicReplicationFactor, - }) - } - if len(missing) == 0 { - return nil + missing, err := b.unknownTopics(names) + if err != nil || len(missing) == 0 { + return err } admin, err := b.adminClient() @@ -68,10 +57,15 @@ func (b *KafkaBus) ensureTopics(ctx context.Context, names ...string) error { cctx, cancel := context.WithTimeout(ctx, topicAdminTimeout) defer cancel() + // Deliberately outside the lock. Two callers racing to create the same + // topic is harmless: the loser is told it already exists, which is the + // steady state after the first call anyway. results, err := admin.CreateTopics(cctx, missing) if err != nil { return fmt.Errorf("eventbus kafka: create topics: %w", err) } + + created := make([]string, 0, len(results)) for _, r := range results { switch r.Error.Code() { case ckf.ErrNoError: @@ -88,14 +82,52 @@ func (b *KafkaBus) ensureTopics(ctx context.Context, names ...string) error { } return fmt.Errorf("eventbus kafka: create topic %q: %s%s", r.Topic, r.Error.String(), hint) } - b.topics.known[r.Topic] = struct{}{} + created = append(created, r.Topic) } + + b.topics.mu.Lock() + for _, t := range created { + b.topics.known[t] = struct{}{} + } + b.topics.mu.Unlock() return nil } +// unknownTopics returns specs for the names this process has not created yet. +// Short critical section: no network happens under the lock. +func (b *KafkaBus) unknownTopics(names []string) ([]ckf.TopicSpecification, error) { + b.topics.mu.Lock() + defer b.topics.mu.Unlock() + if b.topics.closed { + return nil, errors.New("eventbus kafka: bus closed") + } + var missing []ckf.TopicSpecification + for _, n := range names { + if n == "" { + continue + } + if _, seen := b.topics.known[n]; seen { + continue + } + missing = append(missing, ckf.TopicSpecification{ + Topic: n, + NumPartitions: topicPartitions, + ReplicationFactor: topicReplicationFactor, + }) + } + return missing, nil +} + // adminClient opens the admin connection lazily, so a deployment that never // needs to create a topic never opens one. func (b *KafkaBus) adminClient() (*ckf.AdminClient, error) { + b.topics.mu.Lock() + defer b.topics.mu.Unlock() + // Close marks this before it clears the client, so a Publish that raced + // past the closed check cannot open a replacement nothing will shut. + if b.topics.closed { + return nil, errors.New("eventbus kafka: bus closed") + } if b.topics.admin != nil { return b.topics.admin, nil } diff --git a/internal/infrastructure/eventbus/kafka_topics_test.go b/internal/infrastructure/eventbus/kafka_topics_test.go index 1a0c630b..0d443a4d 100644 --- a/internal/infrastructure/eventbus/kafka_topics_test.go +++ b/internal/infrastructure/eventbus/kafka_topics_test.go @@ -5,6 +5,7 @@ package eventbus import ( "context" "testing" + "time" ) // A topic already created by this process must not reach the broker again: @@ -47,3 +48,47 @@ func TestBothConstructorsInitialiseTopicMap(t *testing.T) { t.Fatal("NewKafkaFromProducer left the topic map nil") } } + +// A bus that has been closed must not open a new admin connection. Publish +// checks b.closed and then calls ensureTopics, so a Close landing between the +// two used to resurrect a client that nothing would ever shut. +func TestEnsureTopicsRefusesAfterClose(t *testing.T) { + b := &KafkaBus{ + bootstrap: "localhost:0", + topics: topicEnsurer{known: map[string]struct{}{}, closed: true}, + } + if err := b.ensureTopics(context.Background(), "w.something"); err == nil { + t.Fatal("a closed bus accepted a topic creation") + } + if b.topics.admin != nil { + t.Error("a closed bus opened an admin connection") + } + if _, err := b.adminClient(); err == nil { + t.Error("adminClient handed out a client after close") + } +} + +// The lock must not be held across the broker call, or one slow creation +// stalls Publish, Subscribe and Close for every other topic. Asserted by +// taking the lock and confirming the read-only path still completes. +func TestUnknownTopicsDoesNotHoldLockForCaller(t *testing.T) { + b := &KafkaBus{ + bootstrap: "localhost:0", + topics: topicEnsurer{known: map[string]struct{}{"a": {}}}, + } + missing, err := b.unknownTopics([]string{"a"}) + if err != nil { + t.Fatalf("unknownTopics: %v", err) + } + if len(missing) != 0 { + t.Fatalf("a known topic was reported missing: %v", missing) + } + // The lock is free the moment unknownTopics returns. + done := make(chan struct{}) + go func() { b.topics.mu.Lock(); b.topics.mu.Unlock(); close(done) }() + select { + case <-done: + case <-time.After(2 * time.Second): + t.Fatal("the topic lock was still held after unknownTopics returned") + } +}