Fix task log persistence on Postgres and harden the log flush path

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,

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

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,73 +66,80 @@ 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
 			}
 		}
 
 		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 {

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

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()
 	}
 }