Merge branch 'main' into feature/editor-image-links-and-buttons

.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. 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: |
           go test -race -coverprofile=coverage.out -covermode=atomic ./...

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,11 @@ check-migrations:
 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 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
 # container start, so the same script renders it at build time. This runs both

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 |
 
+<Callout title="Topics are created by the bus, not by hand">
+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.
+</Callout>
+
+
 `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.
 
 <Callout type="warn" title="The tracking topic is read by two languages">

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,16 @@ 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 {
+	// 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
+	}
+	b.topics.mu.Unlock()
+
 	b.mu.Lock()
 	if b.closed {
 		b.mu.Unlock()

internal/infrastructure/eventbus/kafka_topics.go

@@ -0,0 +1,148 @@
+//go:build kafka
+
+package eventbus
+
+import (
+	"context"
+	"errors"
+	"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 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 {
+	missing, err := b.unknownTopics(names)
+	if err != nil || len(missing) == 0 {
+		return err
+	}
+
+	admin, err := b.adminClient()
+	if err != nil {
+		return err
+	}
+
+	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:
+			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)
+		}
+		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
+	}
+	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
+}

internal/infrastructure/eventbus/kafka_topics_test.go

@@ -0,0 +1,94 @@
+//go:build kafka
+
+package eventbus
+
+import (
+	"context"
+	"testing"
+	"time"
+)
+
+// 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")
+	}
+}
+
+// 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")
+	}
+}