fix: add ClickHouse client and InsertMany timeouts (#700)

* chore: improve logmq logs

Add message_id to the handler log entry, event_id, attempt_id, and
tenant_id after parsing in the batch processor. This makes it possible
to trace message redeliveries during pipeline stalls.

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>

* fix: add ClickHouse client and InsertMany timeouts

Add transport-level timeouts (DialTimeout: 10s, ReadTimeout: 30s) to the
ClickHouse client to prevent TCP-level hangs like the Feb 10 incident
where a hung write took 56 min to naturally fail via read: EOF.

Also add a 30s application-level context timeout to the InsertMany call
in the batch processor, bounding how long a single flush can block
regardless of transport behavior.

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>

* chore: log insert_duration_ms in batch processor

Add insert duration timing to both success and error log lines
in processBatch, making it easier to spot slow ClickHouse inserts.

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>

---------

Co-authored-by: Claude Opus 4.6 <noreply@anthropic.com>

Author: alexluong

internal/clickhouse/clickhouse.go

@@ -2,6 +2,7 @@ package clickhouse
 
 import (
 	"crypto/tls"
+	"time"
 
 	"github.com/ClickHouse/clickhouse-go/v2"
 	chdriver "github.com/ClickHouse/clickhouse-go/v2"
@@ -29,6 +30,8 @@ func New(config *ClickHouseConfig) (DB, error) {
 			Username: config.Username,
 			Password: config.Password,
 		},
+		DialTimeout: 10 * time.Second,
+		ReadTimeout: 30 * time.Second,
 
 		// Debug: true,
 		// Debugf: func(format string, v ...any) {

internal/logmq/batchprocessor.go

@@ -90,14 +90,30 @@ func (bp *BatchProcessor) processBatch(_ string, msgs []*mqs.Message) {
 		// Validate that both Event and Attempt are present.
 		// The logstore requires both for data consistency.
 		if entry.Event == nil || entry.Attempt == nil {
-			logger.Error("invalid log entry: both event and attempt are required",
+			fields := []zap.Field{
 				zap.Bool("has_event", entry.Event != nil),
 				zap.Bool("has_attempt", entry.Attempt != nil),
-				zap.String("message_id", msg.LoggableID))
+				zap.String("message_id", msg.LoggableID),
+			}
+			if entry.Event != nil {
+				fields = append(fields, zap.String("event_id", entry.Event.ID))
+				fields = append(fields, zap.String("tenant_id", entry.Event.TenantID))
+			}
+			if entry.Attempt != nil {
+				fields = append(fields, zap.String("attempt_id", entry.Attempt.ID))
+				fields = append(fields, zap.String("tenant_id", entry.Attempt.TenantID))
+			}
+			logger.Error("invalid log entry: both event and attempt are required", fields...)
 			msg.Nack()
 			continue
 		}
 
+		logger.Info("added to batch",
+			zap.String("message_id", msg.LoggableID),
+			zap.String("event_id", entry.Event.ID),
+			zap.String("attempt_id", entry.Attempt.ID),
+			zap.String("tenant_id", entry.Event.TenantID))
+
 		entries = append(entries, entry)
 		validMsgs = append(validMsgs, msg)
 	}
@@ -107,17 +123,24 @@ func (bp *BatchProcessor) processBatch(_ string, msgs []*mqs.Message) {
 		return
 	}
 
-	if err := bp.logStore.InsertMany(bp.ctx, entries); err != nil {
+	insertCtx, cancel := context.WithTimeout(bp.ctx, 30*time.Second)
+	defer cancel()
+
+	insertStart := time.Now()
+	if err := bp.logStore.InsertMany(insertCtx, entries); err != nil {
 		logger.Error("failed to insert log entries",
 			zap.Error(err),
-			zap.Int("entry_count", len(entries)))
+			zap.Int("entry_count", len(entries)),
+			zap.Int64("insert_duration_ms", time.Since(insertStart).Milliseconds()))
 		for _, msg := range validMsgs {
 			msg.Nack()
 		}
 		return
 	}
 
-	logger.Info("batch processed successfully", zap.Int("count", len(validMsgs)))
+	logger.Info("batch processed successfully",
+		zap.Int("count", len(validMsgs)),
+		zap.Int64("insert_duration_ms", time.Since(insertStart).Milliseconds()))
 
 	for _, msg := range validMsgs {
 		msg.Ack()

internal/logmq/messagehandler.go

@@ -6,6 +6,7 @@ import (
 	"github.com/hookdeck/outpost/internal/consumer"
 	"github.com/hookdeck/outpost/internal/logging"
 	"github.com/hookdeck/outpost/internal/mqs"
+	"go.uber.org/zap"
 )
 
 // BatchAdder is the interface for adding messages to a batch processor.
@@ -29,7 +30,8 @@ func NewMessageHandler(logger *logging.Logger, batchAdder BatchAdder) consumer.M
 
 func (h *messageHandler) Handle(ctx context.Context, msg *mqs.Message) error {
 	logger := h.logger.Ctx(ctx)
-	logger.Info("logmq handler")
+	logger.Info("logmq handler",
+		zap.String("message_id", msg.LoggableID))
 	h.batchAdder.Add(ctx, msg)
 	return nil
 }