internal/database/testing.go

@@ -24,11 +24,24 @@ func NewTestDatabase(db *gorm.DB) *Database {
 // NewTestWebhookDBManager creates a WebhookDBManager backed by the given
 // data directory. Intended for use in tests without the fx lifecycle.
 func NewTestWebhookDBManager(dataDir string) *WebhookDBManager {
-	return &WebhookDBManager{
-		dataDir: dataDir,
-		log: slog.New(slog.NewTextHandler(
+	return NewTestWebhookDBManagerWithLogger(
+		dataDir,
+		slog.New(slog.NewTextHandler(
 			os.Stderr,
 			&slog.HandlerOptions{Level: slog.LevelDebug},
 		)),
+	)
+}
+
+// NewTestWebhookDBManagerWithLogger is NewTestWebhookDBManager with the
+// logger supplied by the caller. The per-webhook databases this manager
+// opens hand that logger to gormlog, so a test that needs to see the SQL
+// the service emits can capture it.
+func NewTestWebhookDBManagerWithLogger(
+	dataDir string, log *slog.Logger,
+) *WebhookDBManager {
+	return &WebhookDBManager{
+		dataDir: dataDir,
+		log:     log,
 	}
 }

internal/delivery/queue_depth.go

@@ -95,6 +95,8 @@ func (e *Engine) sampleQueueDepths(ctx context.Context) {
 // targetTypesByID maps every configured target id to its type. The
 // deliveries live in the per-webhook databases but carry only a
 // target id, so the type label has to come from the main database.
+//
+// Find rather than Scan: see sampleWebhookQueueDepths.
 func (e *Engine) targetTypesByID() (
 	map[string]database.TargetType, error,
 ) {
@@ -106,7 +108,7 @@ func (e *Engine) targetTypesByID() (
 	err := e.database.DB().
 		Model(&database.Target{}).
 		Select("id", "type").
-		Scan(&rows).Error
+		Find(&rows).Error
 	if err != nil {
 		return nil, fmt.Errorf("loading targets: %w", err)
 	}
@@ -128,6 +130,13 @@ func (e *Engine) targetTypesByID() (
 // folds that into the unknown series rather than dropping it: a
 // backlog stuck behind a deleted target is a backlog that still needs
 // to be alertable.
+//
+// The aggregate is read with Find, not Scan. (*gorm.DB).Scan swaps
+// GORM's own trace recorder in for the logging adapter, and that
+// recorder does not implement gorm.ParamsFilter, so the statement
+// reaches the log with its bound values interpolated — here, the
+// status list. Find goes through the normal query callback, which is
+// filtered. See internal/gormlog and its scan_guard_test.go.
 func (e *Engine) sampleWebhookQueueDepths(
 	webhookID string,
 	types map[string]database.TargetType,
@@ -158,7 +167,7 @@ func (e *Engine) sampleWebhookQueueDepths(
 			database.DeliveryStatusRetrying,
 		}).
 		Group("target_id, status").
-		Scan(&rows).Error
+		Find(&rows).Error
 	if err != nil {
 		e.log.Error(
 			"queue depth sample: "+

internal/delivery/queue_depth_gormlog_test.go

@@ -0,0 +1,179 @@
+package delivery_test
+
+import (
+	"bytes"
+	"context"
+	"database/sql"
+	"fmt"
+	"log/slog"
+	"net/http"
+	"path/filepath"
+	"strings"
+	"sync"
+	"testing"
+	"time"
+
+	"github.com/google/uuid"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+	"gorm.io/driver/sqlite"
+	"gorm.io/gorm"
+	"sneak.berlin/go/webhooker/internal/database"
+	"sneak.berlin/go/webhooker/internal/delivery"
+	"sneak.berlin/go/webhooker/internal/gormlog"
+)
+
+// qdAggregateMarker identifies the queue-depth aggregate in the
+// captured SQL. It is the one statement in this test that binds
+// anything, and the raw count() expression appears in no other.
+const qdAggregateMarker = "count(*)"
+
+// qdSyncBuf collects log output from whichever goroutine GORM writes
+// on.
+type qdSyncBuf struct {
+	mu sync.Mutex
+	b  bytes.Buffer
+}
+
+func (q *qdSyncBuf) Write(p []byte) (int, error) {
+	q.mu.Lock()
+	defer q.mu.Unlock()
+
+	return q.b.Write(p)
+}
+
+func (q *qdSyncBuf) String() string {
+	q.mu.Lock()
+	defer q.mu.Unlock()
+
+	return q.b.String()
+}
+
+// qdMainDB opens a main database whose GORM logger is the service's
+// adapter, writing through log.
+func qdMainDB(t *testing.T, log *slog.Logger) *gorm.DB {
+	t.Helper()
+
+	dsn := fmt.Sprintf(
+		"file:%s?cache=shared&mode=rwc",
+		filepath.Join(t.TempDir(), "main-gormlog.db"),
+	)
+
+	sqlDB, err := sql.Open("sqlite", dsn)
+	require.NoError(t, err)
+
+	t.Cleanup(func() { _ = sqlDB.Close() })
+
+	db, err := gorm.Open(
+		sqlite.Dialector{Conn: sqlDB},
+		&gorm.Config{Logger: gormlog.New(log)},
+	)
+	require.NoError(t, err)
+
+	require.NoError(t, db.AutoMigrate(
+		&database.Webhook{},
+		&database.Target{},
+	))
+
+	return db
+}
+
+// qdLinesContaining returns every captured line carrying marker.
+func qdLinesContaining(out, marker string) []string {
+	var found []string
+
+	for line := range strings.SplitSeq(out, "\n") {
+		if strings.Contains(line, marker) {
+			found = append(found, line)
+		}
+	}
+
+	return found
+}
+
+// TestQueueDepthSample_LogsNoBoundValue holds the queue-depth sampler
+// to the values-off property internal/gormlog exists to provide.
+//
+// The aggregate binds the delivery status list. Read with
+// (*gorm.DB).Scan it was logged with those values interpolated, because
+// Scan records the statement through GORM's own traceRecorder, which
+// does not implement gorm.ParamsFilter. Read with Find it goes through
+// the normal query callback and the adapter's filter applies. Restore
+// the Scan call in queue_depth.go and this fails on the status literals
+// below; scan_guard_test.go catches the same regression statically.
+func TestQueueDepthSample_LogsNoBoundValue(t *testing.T) {
+	t.Parallel()
+
+	buf := &qdSyncBuf{}
+	log := slog.New(slog.NewTextHandler(
+		buf, &slog.HandlerOptions{Level: slog.LevelDebug},
+	))
+
+	mainDB := qdMainDB(t, log)
+	dbMgr := database.NewTestWebhookDBManagerWithLogger(
+		t.TempDir(), log,
+	)
+
+	webhookID := uuid.New().String()
+	webhookDB := iSeedWebhookDB(t, dbMgr, webhookID)
+
+	iCreateWebhook(t, mainDB, webhookID, "queue-depth-gormlog")
+
+	targetID := uuid.New().String()
+
+	iCreateTarget(t, mainDB, targetID, webhookID,
+		"queue-depth-gormlog-target", database.TargetTypeHTTP,
+		iHTTPConfig("https://example.com/hook"), 3,
+	)
+
+	event := iSeedEvent(
+		t, webhookDB, webhookID, `{"queued":true}`,
+	)
+
+	iSeedDelivery(
+		t, webhookDB, event.ID, targetID,
+		database.DeliveryStatusPending,
+	)
+	iSeedDelivery(
+		t, webhookDB, event.ID, targetID,
+		database.DeliveryStatusRetrying,
+	)
+
+	engine := delivery.NewTestEngineWithDB(
+		database.NewTestDatabase(mainDB),
+		dbMgr,
+		log,
+		&http.Client{Timeout: 5 * time.Second},
+		2,
+	)
+
+	engine.ExportSampleQueueDepths(context.Background())
+
+	out := buf.String()
+
+	lines := qdLinesContaining(out, qdAggregateMarker)
+	require.NotEmpty(
+		t, lines,
+		"the queue-depth aggregate was never logged, so the "+
+			"assertions below are vacuous",
+	)
+
+	for _, line := range lines {
+		assert.Contains(
+			t, line, "?",
+			"the aggregate was logged without its placeholders: %s",
+			line,
+		)
+
+		for _, status := range []database.DeliveryStatus{
+			database.DeliveryStatusPending,
+			database.DeliveryStatusRetrying,
+		} {
+			assert.NotContains(
+				t, line, string(status),
+				"a bound status value was interpolated into the "+
+					"logged statement: %s", line,
+			)
+		}
+	}
+}