Event-driven Architecture for SmokeRevel: 5 Tips (2026)

I pushed a new build of our smoke‑test suite at 02:13 am, convinced the team that the “fire‑and‑forget” webhook was iron‑clad. By 02:45 am the dashboard was flashing **FAILED** for every branch, and the alerts started pounding Slack. The culprit? A single flaky test that retried, doubled‑publishing its result and blowing up the aggregation logic that assumed one‑to‑one mapping. That nightmare taught me the hard way that pulling results with a periodic REST call or a naïve webhook is a recipe for race conditions, data loss, and endless debugging sessions.

Event‑Driven Architecture for SmokeRevel Test Aggregation in 2026

Event-driven architecture (EDA) uses message brokers like Apache Pulsar to decouple SmokeRevel test execution from result aggregation. In 2026, this enables real-time, scalable processing by publishing test results as events, which are consumed asynchronously to build consolidated reports, dashboards, and trigger downstream actions with high reliability.

—

Why Event-Driven Architecture for Modern Test Aggregation?

Challenges of Direct Polling and REST APIs

Polling a `/results` endpoint every few seconds sounds simple, but in a CI/CD fleet that executes **thousands** of test suites per hour it becomes a bottleneck. Each poll adds latency, consumes extra network bandwidth, and forces the test runner to expose its internal state over a public API—an attack surface you don’t need. Moreover, a synchronous response model can’t guarantee delivery when a runner crashes mid‑execution; the missing payload is lost forever.

Real-Time Scalability Demands for CI/CD

Continuous delivery pipelines today push code to production several times a day. The moment a test fails, engineers expect to see the failure in under a second so they can halt a rollout. An event stream gives you sub‑millisecond propagation: as soon as a test worker publishes a `TestResult` event, all downstream consumers (dashboards, alerting rules, roll‑back orchestrators) see it instantly. The scaling story is straightforward—just spin up more consumer instances in the same consumer group; Pulsar handles partition rebalancing for you.

**My take:** If you’re still relying on a monolithic “results API”, you’re paying a hidden price in latency and operational complexity. Switch to streaming early; the upgrade cost pays for itself the first time you need to double your test throughput.

Core Components of an Event-Driven Test Aggregation Pipeline

The Message Broker (Apache Pulsar vs. Kafka)

Both Pulsar 3.0+ and Kafka 3.5+ are mature, but Pulsar gives us multi‑tenant isolation out of the box and a **tiered storage** model that drops old partitions to S3 without admin churn. Confluent Cloud’s 2025 schema registry also offers a nice UI for Avro/Protobuf evolution, but it lacks Pulsar’s per‑tenant throttling. In practice, I run a Pulsar cluster on EKS with three broker pods per zone, each backed by a 3‑node BookKeeper ensemble.

Feature	Pulsar 3.0+	Kafka 3.5+
Multi‑tenant isolation	Native namespace quotas	Requires separate clusters or ACLs
Tiered storage	Built‑in S3 tiering	MirrorMaker 2 or external connectors
Exactly‑once support	End‑to‑end with Pulsar Functions (beta)	Transactional producer API
Ops overhead	Lower (single binary)	Higher (zookeeper + brokers)

Event Model Design for SmokeRevel Results

The event schema lives in a protobuf file (`test_result.proto`) so we can evolve it without breaking old consumers. A minimal version looks like this:

// test_result.proto - version 1
syntax = "proto3";

package smokerevel;

message TestResult {
  string test_id = 1;           // UUID from SmokeRevel
  string suite_id = 2;          // Grouping ID
  string status = 3;            // PASS | FAIL | SKIP
  int64  duration_ms = 4;
  string run_at = 5;            // RFC3339 timestamp
  map<string, string> metadata = 6;
}

When you add a new field, just bump the `proto` version and register the new schema in Pulsar’s schema registry. Our internal guide on **[Protobuf schema evolution]** (link to the tutorial) walks you through backward‑compatible changes.

Aggregator Services and Serverless Functions

Two patterns work side‑by‑side:

1. **Stateful aggregator microservice** – a Go service that maintains an in‑memory map of suite IDs → aggregation structs. It writes snapshots to DynamoDB every 30 seconds for durability.
2. **Serverless fan‑out** – an AWS Lambda (or Pulsar Function) that reacts to each `TestResult` and pushes a flattened row into a ClickHouse table for analytic dashboards.

The hybrid approach lets you keep low‑latency alerts in the microservice while still offering ad‑hoc querying via the warehouse.

Step-by-Step Implementation with 2024‑2026 SDKs and Tools

Configuring SmokeRevel’s Event Producer

SmokeRevel 4.2+ ships a Go client that can publish directly to a Pulsar topic. First, install the SDK:

go get github.com/smokerevel/sdk/v4@v4.2.1
go get github.com/apache/pulsar-client-go/pulsar@v3.0.0

// producer.go – Go 1.23
package main

import (
	"context"
	"log"
	"time"

	sr "github.com/smokerevel/sdk/v4"
	"github.com/apache/pulsar-client-go/pulsar"
	"google.golang.org/protobuf/proto"
	pb "myorg/tests/proto"
)

func main() {
	// 1️⃣ Create Pulsar client (replicated across three zones)
	client, err := pulsar.NewClient(pulsar.ClientOptions{
		URL:               "pulsar+ssl://pulsar-us-east-1.mycompany.com:6651",
		Authentication:    pulsar.NewAuthenticationToken("{{PULSAR_TOKEN}}"),
		OperationTimeout:  30 * time.Second,
		ConnectionTimeout: 10 * time.Second,
	})
	if err != nil {
		log.Fatalf("pulsar client init failed: %v", err)
	}
	defer client.Close()

	// 2️⃣ Configure producer with schema enforcement
	producer, err := client.CreateProducer(pulsar.ProducerOptions{
		Topic:           "persistent://public/default/smokerevel-results",
		EnableBatching:  true,
		BatchMaxMessages: 500,
		SendTimeout:     10 * time.Second,
	})
	if err != nil {
		log.Fatalf("producer creation failed: %v", err)
	}
	defer producer.Close()

	// 3️⃣ Wire SmokeRevel SDK to emit events
	srClient := sr.NewClient(sr.Config{
		Token:   "{{SMOKEREVEL_TOKEN}}",
		BaseURL: "https://api.smokerevel.com",
	})

	// Register a callback that runs after each test finishes
	srClient.OnResult(func(res sr.Result) {
		ev := &pb.TestResult{
			TestId:     res.ID,
			SuiteId:    res.SuiteID,
			Status:     string(res.Status),
			DurationMs: int64(res.Duration / time.Millisecond),
			RunAt:      res.StartTime.Format(time.RFC3339),
			Metadata:   res.Labels,
		}
		data, err := proto.Marshal(ev)
		if err != nil {
			log.Printf("proto marshal failed: %v", err)
			return
		}
		// 4️⃣ Publish with retry backoff
		sendWithRetry(context.Background(), producer, data)
	})

	// Block forever – the SDK runs its own internal scheduler
	select {}
}

// sendWithRetry publishes a message and retries on transient errors.
// Exponential backoff caps at 5 attempts.
func sendWithRetry(ctx context.Context, p pulsar.Producer, payload []byte) {
	const maxAttempts = 5
	var attempt int
	for {
		_, err := p.Send(ctx, &pulsar.Message{
			Payload: payload,
		})
		if err == nil {
			return
		}
		attempt++
		if attempt >= maxAttempts {
			log.Printf("max retries reached, dropping message: %v", err)
			return
		}
		backoff := time.Duration(attempt*200) * time.Millisecond
		log.Printf("publish retry %d after %v: %v", attempt, backoff, err)
		time.Sleep(backoff)
	}
}

Key points:

• **Durable producer** – we enable batching but set a short `SendTimeout` so the client doesn’t hang forever.
• **Retry logic** – exponential backoff protects us from temporary network glitches.
• **Idempotency** – each `TestResult` contains a stable `test_id`; downstream consumers can dedupe on that key.

Building the Result Consumer with Go 1.23

A consumer must be able to replay messages after a crash, so we store the latest processed `test_id` in DynamoDB. Here’s a minimalist, production‑ready consumer:

// consumer.go – Go 1.23
package main

import (
	"context"
	"encoding/json"
	"log"
	"time"

	"github.com/apache/pulsar-client-go/pulsar"
	"github.com/aws/aws-sdk-go-v2/config"
	"github.com/aws/aws-sdk-go-v2/service/dynamodb"
	pb "myorg/tests/proto"
	"google.golang.org/protobuf/proto"
)

const (
	topic          = "persistent://public/default/smokerevel-results"
	subscription   = "aggregator-group"
	stateTableName = "SmokeRevelAggState"
)

type AggState struct {
	SuiteID   string
	PassCount int64
	FailCount int64
	SkipCount int64
	TotalDur  int64 // ms
	UpdatedAt int64 // epoch seconds
}

// DynamoDB helper – upsert aggregation state
func upsertState(ctx context.Context, db *dynamodb.Client, state AggState) error {
	item, err := json.Marshal(state)
	if err != nil {
		return err
	}
	_, err = db.PutItem(ctx, &dynamodb.PutItemInput{
		TableName: &stateTableName,
		Item: map[string]dynamodb.AttributeValue{
			"suite_id":    {S: &state.SuiteID},
			"payload":     {S: (*string)(item)},
			"last_updated": {N: awsString(fmt.Sprintf("%d", state.UpdatedAt))},
		},
	})
	return err
}

// Main consumer loop
func main() {
	// 1️⃣ Pulsar client
	client, err := pulsar.NewClient(pulsar.ClientOptions{
		URL: "pulsar+ssl://pulsar-us-east-1.mycompany.com:6651",
		Authentication: pulsar.NewAuthenticationToken("{{PULSAR_TOKEN}}"),
	})
	if err != nil {
		log.Fatalf("pulsar client error: %v", err)
	}
	defer client.Close()

	// 2️⃣ AWS SDK for DynamoDB
	awsCfg, err := config.LoadDefaultConfig(context.Background())
	if err != nil {
		log.Fatalf("aws config load failed: %v", err)
	}
	db := dynamodb.NewFromConfig(awsCfg)

	// 3️⃣ Consumer with a shared subscription (fan‑out)
	consumer, err := client.Subscribe(pulsar.ConsumerOptions{
		Topic:            topic,
		SubscriptionName: subscription,
		Type:             pulsar.Shared,
	})
	if err != nil {
		log.Fatalf("consumer subscribe failed: %v", err)
	}
	defer consumer.Close()

	ctx := context.Background()
	for {
		msg, err := consumer.Receive(ctx)
		if err != nil {
			log.Printf("receive error: %v", err)
			continue
		}
		go handleMessage(ctx, consumer, msg, db)
	}
}

// handleMessage processes a single TestResult, dedupes, and updates state.
func handleMessage(ctx context.Context, consumer pulsar.Consumer, msg pulsar.Message, db *dynamodb.Client) {
	defer consumer.Ack(msg) // Acknowledge regardless of outcome to avoid redelivery loops

	var tr pb.TestResult
	if err := proto.Unmarshal(msg.Payload(), &tr); err != nil {
		log.Printf("proto unmarshal failed: %v", err)
		return
	}

	// Simple idempotency check – use DynamoDB conditional write
	// (skip if this test_id already exists). In production we'd have a separate
	// “processed_ids” table with TTL.
	// For brevity we assume the check succeeded.

	// Update in‑memory aggregation (could be a Redis cache)
	state := AggState{
		SuiteID:   tr.SuiteId,
		PassCount: 0,
		FailCount: 0,
		SkipCount: 0,
		TotalDur:  0,
		UpdatedAt: time.Now().Unix(),
	}
	switch tr.Status {
	case "PASS":
		state.PassCount = 1
	case "FAIL":
		state.FailCount = 1
	case "SKIP":
		state.SkipCount = 1
	}
	state.TotalDur = tr.DurationMs

	if err := upsertState(ctx, db, state); err != nil {
		log.Printf("state upsert failed: %v", err)
		// We *don't* Ack here; the message will be redelivered after the