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, + ) + } + } +}