Fix bug in kafka implementation that caused double client registration

.github/workflows/release_build.yml

@@ -55,11 +55,13 @@ jobs:
         env:
           GITHUB_TOKEN: ${{ secrets.GITHUB_TOKEN }}
 
-      - name: Run GoReleaser (dev snapshot)
+      - name: Run GoReleaser (dev)
         if: github.ref == 'refs/heads/dev'
         uses: goreleaser/goreleaser-action@v7
         with:
           version: latest
-          args: release --snapshot --clean --config .github/goreleaser.yml
+          args: release --clean --config .github/goreleaser.yml --skip=validate
         env:
+          GORELEASER_CURRENT_TAG: dev
+
           GITHUB_TOKEN: ${{ secrets.GITHUB_TOKEN }}

CHANGELOG.md

@@ -15,6 +15,12 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ### Fixed
 
+## [1.10.1] - 2026-09-13
+
+### Fixed
+
+- Prevent duplicate client registration by checking if the client is already registered before adding it (#116)
+
 ## [1.10.0] - 2026-09-13
 
 ### Added
@@ -30,16 +36,20 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 - Updated http client settings to prevent timeouts and other connectivity issues
 - Updated http server settings to allow for higher delays
 - Updated batch size from 100 to 256 (#97)
+- Use config.yaml as default value for --config arg (6aed28e)
 
 ### Fixed
 
 - Prevent excessive tile fetching for static CT logs (#104)
 - Fix off-by-one is Static CT path causing refetches of non-growing logs by @FiloSottile in https://github.com/d-Rickyy-b/certstream-server-go/pull/114
-- Calculate correct starting tile index for static ct logs (a2b4a9f)
 - Use proper websocket close code (1008) instead of 1005, which wasn't sent to the client (76ff240)
+- Properly handle websocket client disconnections (750ed24)
 - Respect ct_index path in config file for the `create-index` command (67838a9)
 - Ensure all additionalLog entries are processed correctly (c8bd735)
 - Reload CT index from metrics before restarting the worker (85e90bf, #113)
+- Use double dashes for docker entrypoint --config argument (#101)
+- Use ct_index path from config file if no outfile is given (#102)
+- Close temp file before renaming it (fb10013)
 
 ## [1.9.0] - 2026-04-03
 
@@ -251,7 +261,8 @@ Fixed the connection issue due to the offline Google loglist urls.
 
 Initial release! First stable version of certstream-server-go is published as v1.0.0
 
-[unreleased]: https://github.com/d-Rickyy-b/certstream-server-go/compare/v1.10.0...HEAD
+[unreleased]: https://github.com/d-Rickyy-b/certstream-server-go/compare/v1.10.1...HEAD
+[1.10.1]: https://github.com/d-Rickyy-b/certstream-server-go/compare/v1.10.0...v1.10.1
 [1.10.0]: https://github.com/d-Rickyy-b/certstream-server-go/compare/v1.9.0...v1.10.0
 [1.9.0]: https://github.com/d-Rickyy-b/certstream-server-go/compare/v1.8.2...v1.9.0
 [1.8.2]: https://github.com/d-Rickyy-b/certstream-server-go/compare/v1.8.1...v1.8.2

internal/broadcast/dispatcher.go

@@ -33,12 +33,19 @@ func NewDispatcher() *Dispatcher {
 // RegisterClient adds a client to the list of clients of the Dispatcher.
 // The client will receive certificate broadcasts right after registration.
 func (bm *Dispatcher) RegisterClient(c CertProcessor) {
-	// TODO: check if the client is already registered
 	bm.clientLock.Lock()
+	defer bm.clientLock.Unlock()
+
+	for _, client := range bm.clients {
+		if client == c {
+			log.Printf("Client already registered: %s\n", c.Name())
+			return
+		}
+	}
+
 	bm.clients = append(bm.clients, c)
 	log.Printf("Added new client. Clients: %d, Capacity: %d\n", len(bm.clients), cap(bm.clients))
 	metrics.Prometheus.RegisterClient(c.Name(), func() float64 { return float64(c.SkippedCerts()) })
-	bm.clientLock.Unlock()
 }
 
 // UnregisterClient removes a client from the list of clients of the Dispatcher.

internal/broadcast/dispatcher_test.go

@@ -14,6 +14,18 @@ func newDispatcherTestWebsocketClient(name string) *WebsocketClient {
 	}
 }
 
+func TestDispatcherRegisterClientTwice(t *testing.T) {
+	d := &Dispatcher{}
+	client := newDispatcherTestWebsocketClient("client-1")
+
+	d.RegisterClient(client)
+	d.RegisterClient(client)
+
+	if got := len(d.clients); got != 1 {
+		t.Fatalf("expected 1 registered client after duplicate register, got %d", got)
+	}
+}
+
 func TestDispatcherUnregisterWebsocketClient(t *testing.T) {
 	d := &Dispatcher{}
 	client := newDispatcherTestWebsocketClient("client-1")

internal/certstream/certstream.go

@@ -96,8 +96,6 @@ func NewCertstreamServer(cfg config.Config) (*Certstream, error) {
 				string(streamProcessor.Compression),
 				cfg.General.BufferSizes.Websocket,
 			)
-
-			broadcast.ClientHandler.RegisterClient(c)
 		default:
 			log.Printf("Unknown stream processor type '%s' for %s. Skipping...\n", streamProcessor.Type, streamProcessor.Name)
 		}

internal/config/config.go

@@ -13,7 +13,7 @@ import (
 var (
 	// AppConfig holds the parsed configuration.
 	AppConfig Config
-	Version   = "1.10.0"
+	Version   = "1.10.1"
 
 	ErrInvalidConfig = errors.New("invalid configuration")
 	URLPathRegex     = regexp.MustCompile(`^(/[a-zA-Z0-9\-._]+)+$`)