From 354b271d359a01c9635fbcc7234b0c18027660de Mon Sep 17 00:00:00 2001 From: clawbot Date: Thu, 20 Aug 2026 05:41:55 +0000 Subject: [PATCH] Read queue depths with Find, not Scan (closes #234) (*gorm.DB).Scan swaps GORM's own traceRecorder in for the configured logger for the duration of the statement, and that recorder does not implement gorm.ParamsFilter. The statement therefore reaches the log with its bound values interpolated, which is the one path (*gormlog.Logger).ParamsFilter cannot reach. internal/gormlog's scan_guard_test.go exists to keep that path out of production code; the queue-depth sampler landed with two calls on it, so next has been failing make check on its own. Both call sites now use Find, which goes through the normal query callback. The emitted SQL is otherwise unchanged -- callbacks.Query and callbacks.RowQuery share BuildQuerySQL, and both call sites set Model and Select explicitly, so the table, the column list and the soft-delete clause are built identically. Only the log line differs: Scan: ... WHERE status IN ("pending","retrying") AND ... Find: ... WHERE status IN (?,?) AND ... TestQueueDepthSample_LogsNoBoundValue drives one sample through the adapter and asserts the aggregate keeps its placeholders and carries no status literal. Restoring either Scan fails it as well as the static guard. database.NewTestWebhookDBManagerWithLogger lets that test capture the SQL the per-webhook databases emit; NewTestWebhookDBManager keeps its signature and delegates to it. --- internal/database/testing.go | 19 +- internal/delivery/queue_depth.go | 13 +- internal/delivery/queue_depth_gormlog_test.go | 179 ++++++++++++++++++ 3 files changed, 206 insertions(+), 5 deletions(-) create mode 100644 internal/delivery/queue_depth_gormlog_test.go diff --git a/internal/database/testing.go b/internal/database/testing.go index 6327eac..72c8816 100644 --- a/internal/database/testing.go +++ b/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, } } diff --git a/internal/delivery/queue_depth.go b/internal/delivery/queue_depth.go index 760a5a6..e627650 100644 --- a/internal/delivery/queue_depth.go +++ b/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: "+ diff --git a/internal/delivery/queue_depth_gormlog_test.go b/internal/delivery/queue_depth_gormlog_test.go new file mode 100644 index 0000000..e733990 --- /dev/null +++ b/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, + ) + } + } +}