From fc754e01ac9ce995cf868f3b70caec5d41d186d5 Mon Sep 17 00:00:00 2001 From: Damilola Edwards Date: Sat, 11 Jul 2026 09:33:52 +0100 Subject: [PATCH] Fix task log persistence on Postgres and harden the log flush path The task log upsert used ON CONFLICT (run_id, task_id, log_time), but the only unique key on task_logs is the primary key (run_id, task_id, log_idx). Postgres rejects an ON CONFLICT target that does not match a unique constraint, so every task log insert failed on Postgres and no task logs were stored. Point the conflict target at the primary key, which both engines share. The flush path also failed badly whenever a flush errored, which on Postgres was every time. On error it logged through the writer's own logger, whose database hook re-entered the flush and deadlocked on the buffer mutex; the buffer was only cleared on success, so it grew without bound; and the flush used a shared channel that could be closed twice and panic. Rewrite the flush coordination around the buffer mutex and a scheduled flag. The buffer is always cleared after a flush attempt, flush errors are reported through the parent logger so they cannot re-enter the hook, and the cancellation channel is removed, so a delayed flush simply finds an empty buffer when an earlier flush already ran. Add tests for persistence and upsert, for progress and a bounded buffer under a failing database, and for the normal persistence path. --- pkg/db/task_logs.go | 2 +- pkg/db/task_logs_test.go | 75 ++++++++++++++++++++++++ pkg/logger/dbwriter.go | 86 +++++++++++++++------------- pkg/logger/dbwriter_test.go | 111 ++++++++++++++++++++++++++++++++++++ pkg/logger/logscope.go | 2 +- 5 files changed, 233 insertions(+), 43 deletions(-) create mode 100644 pkg/db/task_logs_test.go create mode 100644 pkg/logger/dbwriter_test.go diff --git a/pkg/db/task_logs.go b/pkg/db/task_logs.go index 16d600ff..f95ebffc 100644 --- a/pkg/db/task_logs.go +++ b/pkg/db/task_logs.go @@ -18,7 +18,7 @@ func (db *Database) InsertTaskLog(tx *sqlx.Tx, log *TaskLog) error { INSERT INTO task_logs ( run_id, task_id, log_idx, log_time, log_level, log_fields, log_message ) VALUES ($1, $2, $3, $4, $5, $6, $7) - ON CONFLICT (run_id, task_id, log_time) DO UPDATE SET + ON CONFLICT (run_id, task_id, log_idx) DO UPDATE SET log_time = excluded.log_time, log_level = excluded.log_level, log_fields = excluded.log_fields, diff --git a/pkg/db/task_logs_test.go b/pkg/db/task_logs_test.go new file mode 100644 index 00000000..67da1487 --- /dev/null +++ b/pkg/db/task_logs_test.go @@ -0,0 +1,75 @@ +package db + +import ( + "io" + "testing" + + "github.com/jmoiron/sqlx" + "github.com/sirupsen/logrus" +) + +func newSqliteTestDB(t *testing.T) *Database { + t.Helper() + + quiet := logrus.New() + quiet.SetOutput(io.Discard) + + database := NewDatabase(quiet) + if err := database.InitDB(&DatabaseConfig{ + Engine: "sqlite", + Sqlite: &SqliteDatabaseConfig{File: t.TempDir() + "/test.sqlite"}, + }); err != nil { + t.Fatalf("init db: %v", err) + } + + if err := database.ApplySchema(-2); err != nil { + t.Fatalf("apply schema: %v", err) + } + + return database +} + +// TestInsertTaskLogPersistsAndUpserts verifies that a task log is persisted and +// that re-inserting the same primary key (run_id, task_id, log_idx) updates the +// row in place rather than duplicating or failing. The upsert conflict target +// must be the primary key; any other target is rejected by Postgres at execution +// time, which previously made every insert fail on that engine. +func TestInsertTaskLogPersistsAndUpserts(t *testing.T) { + database := newSqliteTestDB(t) + + insert := func(l *TaskLog) { + if err := database.RunTransaction(func(tx *sqlx.Tx) error { + return database.InsertTaskLog(tx, l) + }); err != nil { + t.Fatalf("insert task log: %v", err) + } + } + + insert(&TaskLog{RunID: 1, TaskID: 1, LogIndex: 1, LogTime: 100, LogLevel: 1, LogMessage: "hello"}) + insert(&TaskLog{RunID: 1, TaskID: 1, LogIndex: 2, LogTime: 110, LogLevel: 1, LogMessage: "world"}) + + logs, err := database.GetTaskLogs(1, 1, 0, 100) + if err != nil { + t.Fatalf("get task logs: %v", err) + } + + if len(logs) != 2 { + t.Fatalf("expected 2 persisted logs, got %d", len(logs)) + } + + // re-insert log index 1 with new content + insert(&TaskLog{RunID: 1, TaskID: 1, LogIndex: 1, LogTime: 200, LogLevel: 2, LogMessage: "updated"}) + + logs, err = database.GetTaskLogs(1, 1, 0, 100) + if err != nil { + t.Fatalf("get task logs: %v", err) + } + + if len(logs) != 2 { + t.Fatalf("after upsert: expected 2 logs, got %d", len(logs)) + } + + if logs[0].LogMessage != "updated" { + t.Fatalf("after upsert: expected log index 1 to read \"updated\", got %q", logs[0].LogMessage) + } +} diff --git a/pkg/logger/dbwriter.go b/pkg/logger/dbwriter.go index 28669bc3..11bec3bb 100644 --- a/pkg/logger/dbwriter.go +++ b/pkg/logger/dbwriter.go @@ -16,14 +16,11 @@ type logDBWriter struct { bufferSize uint64 flushDelay time.Duration - lastIdx uint64 - flushIdx uint64 - bufIdx uint64 - bufMtx sync.Mutex - buf []*db.TaskLog - flushMtx sync.Mutex - flushing bool - flushingChan chan bool + lastIdx uint64 + bufIdx uint64 + bufMtx sync.Mutex + buf []*db.TaskLog + flushScheduled bool } func newLogDBWriter(logger *LogScope, bufferSize uint64, flushDelay time.Duration) *logDBWriter { @@ -44,7 +41,7 @@ func (lh *logDBWriter) Fire(entry *logrus.Entry) error { defer lh.bufMtx.Unlock() if lh.bufIdx >= lh.bufferSize { - lh.flushToDB() + lh.flushToDBLocked() } logIdx := lh.lastIdx + 1 @@ -69,59 +66,59 @@ func (lh *logDBWriter) Fire(entry *logrus.Entry) error { lh.buf = append(lh.buf, taskLog) lh.bufIdx++ - lh.flushDelayed() + lh.scheduleFlushLocked() return nil } -func (lh *logDBWriter) flushDelayed() { - if lh.flushing { +// scheduleFlushLocked arranges for the buffer to be flushed after flushDelay. +// The caller must hold bufMtx. At most one delayed flush is pending at a time; +// a flush triggered earlier by the buffer filling up simply leaves the pending +// timer to find an empty buffer and do nothing. +func (lh *logDBWriter) scheduleFlushLocked() { + if lh.flushScheduled { return } - lh.flushing = true - lh.flushingChan = make(chan bool) + lh.flushScheduled = true - go func() { - defer func() { - lh.flushing = false - }() - - select { - case <-lh.flushingChan: - lh.flushingChan = nil - return - case <-time.After(2 * time.Second): - } + delay := lh.flushDelay + if delay <= 0 { + delay = 2 * time.Second + } - lh.flushingChan = nil + go func() { + time.Sleep(delay) lh.bufMtx.Lock() defer lh.bufMtx.Unlock() - lh.flushToDB() + lh.flushScheduled = false + lh.flushToDBLocked() }() } -func (lh *logDBWriter) flushToDB() { - lh.flushMtx.Lock() - - defer func() { - lh.flushMtx.Unlock() - }() +// flush writes any buffered entries to the database. +func (lh *logDBWriter) flush() { + lh.bufMtx.Lock() + defer lh.bufMtx.Unlock() - if flushingChan := lh.flushingChan; flushingChan != nil { - close(flushingChan) - } + lh.flushToDBLocked() +} +// flushToDBLocked writes the buffered entries to the database. The caller must +// hold bufMtx. The buffer is always cleared afterwards, even on error, so a +// persistent database failure cannot grow it without bound. Errors are reported +// through the parent logger, not this writer's own logger, whose db hook would +// re-enter Fire and deadlock on bufMtx. +func (lh *logDBWriter) flushToDBLocked() { if len(lh.buf) == 0 { return } err := lh.logger.options.Database.RunTransaction(func(tx *sqlx.Tx) error { for _, entry := range lh.buf { - err := lh.logger.options.Database.InsertTaskLog(tx, entry) - if err != nil { + if err := lh.logger.options.Database.InsertTaskLog(tx, entry); err != nil { return err } } @@ -129,13 +126,20 @@ func (lh *logDBWriter) flushToDB() { return nil }) if err != nil { - lh.logger.logger.Errorf("failed to write log entries to db: %v", err) - return + lh.logFlushError(err) } lh.buf = lh.buf[:0] lh.bufIdx = 0 - lh.flushIdx = lh.lastIdx +} + +func (lh *logDBWriter) logFlushError(err error) { + if lh.logger.parentLogger != nil { + lh.logger.parentLogger.WithError(err).Error("failed to write task log entries to db") + return + } + + logrus.WithError(err).Error("failed to write task log entries to db") } func (lh *logDBWriter) getBufferEntries() []*db.TaskLog { diff --git a/pkg/logger/dbwriter_test.go b/pkg/logger/dbwriter_test.go new file mode 100644 index 00000000..5453fffa --- /dev/null +++ b/pkg/logger/dbwriter_test.go @@ -0,0 +1,111 @@ +package logger + +import ( + "io" + "testing" + "time" + + "github.com/ethpandaops/assertoor/pkg/db" + "github.com/sirupsen/logrus" +) + +func newSqliteDB(t *testing.T, applySchema bool) *db.Database { + t.Helper() + + quiet := logrus.New() + quiet.SetOutput(io.Discard) + + database := db.NewDatabase(quiet) + if err := database.InitDB(&db.DatabaseConfig{ + Engine: "sqlite", + Sqlite: &db.SqliteDatabaseConfig{File: t.TempDir() + "/logger.sqlite"}, + }); err != nil { + t.Fatalf("init db: %v", err) + } + + if applySchema { + if err := database.ApplySchema(-2); err != nil { + t.Fatalf("apply schema: %v", err) + } + } + + return database +} + +// TestFlushDoesNotDeadlockOrGrowOnDBError drives the real logger with a database +// whose task_logs table does not exist, so every flush fails. Previously the flush +// error was logged through the writer's own logger, whose db hook re-entered Fire +// and deadlocked on the buffer mutex, and the buffer was never trimmed on error so +// it grew without bound. Both must be gone: logging must make progress and the +// buffer must stay bounded. +func TestFlushDoesNotDeadlockOrGrowOnDBError(t *testing.T) { + quiet := logrus.New() + quiet.SetOutput(io.Discard) + + database := newSqliteDB(t, false) // no schema -> every InsertTaskLog fails + + ls := NewLogger(&ScopeOptions{ + Parent: quiet, + Database: database, + BufferSize: 4, + TestRunID: 1, + TaskID: 1, + }) + log := ls.GetLogger() + + done := make(chan struct{}) + + go func() { + for i := 0; i < 1000; i++ { + log.Infof("entry %d", i) + } + + close(done) + }() + + select { + case <-done: + case <-time.After(15 * time.Second): + t.Fatal("logging deadlocked against a failing database") + } + + ls.dbWriter.bufMtx.Lock() + buffered := len(ls.dbWriter.buf) + ls.dbWriter.bufMtx.Unlock() + + if uint64(buffered) > ls.dbWriter.bufferSize { + t.Fatalf("buffer grew to %d entries on persistent db error (bufferSize %d)", buffered, ls.dbWriter.bufferSize) + } +} + +// TestLogsPersistThroughWriter verifies the happy path still works after the flush +// rewrite: entries logged through the scope are written to the database on flush. +func TestLogsPersistThroughWriter(t *testing.T) { + quiet := logrus.New() + quiet.SetOutput(io.Discard) + + database := newSqliteDB(t, true) + + ls := NewLogger(&ScopeOptions{ + Parent: quiet, + Database: database, + TestRunID: 7, + TaskID: 3, + }) + log := ls.GetLogger() + + log.Info("first") + log.Info("second") + log.Info("third") + + ls.Flush() + + logs, err := database.GetTaskLogs(7, 3, 0, 100) + if err != nil { + t.Fatalf("get task logs: %v", err) + } + + if len(logs) != 3 { + t.Fatalf("expected 3 persisted logs, got %d", len(logs)) + } +} diff --git a/pkg/logger/logscope.go b/pkg/logger/logscope.go index 19437438..7635bf53 100644 --- a/pkg/logger/logscope.go +++ b/pkg/logger/logscope.go @@ -98,7 +98,7 @@ func (ls *LogScope) GetLogger() *logrus.Logger { func (ls *LogScope) Flush() { if ls.dbWriter != nil { - ls.dbWriter.flushToDB() + ls.dbWriter.flush() } }