HACK: Track goroutine IDs to determine when we write by the wrong thread

To use: set `DENDRITE_TRACE_SQL=1` then grep for `unsafe`

internal/sqlutil/trace.go

@@ -31,16 +31,33 @@ import (
 )
 
 var tracingEnabled = os.Getenv("DENDRITE_TRACE_SQL") == "1"
+var dbToWriter map[string]Writer
+var CtxDBInstance = "db_instance"
+var instCount = 0
 
 type traceInterceptor struct {
 	sqlmw.NullInterceptor
+	conn driver.Conn
 }
 
 func (in *traceInterceptor) StmtQueryContext(ctx context.Context, stmt driver.StmtQueryContext, query string, args []driver.NamedValue) (driver.Rows, error) {
 	startedAt := time.Now()
 	rows, err := stmt.QueryContext(ctx, args)
+	key := ctx.Value(CtxDBInstance)
+	var safe string
+	if key != nil {
+		w := dbToWriter[key.(string)]
+		if w == nil {
+			safe = fmt.Sprintf("no writer for key %s", key)
+		} else {
+			safe = w.Safe()
+		}
+	}
+	if safe != "" {
+		logrus.Infof("unsafe: %s -- %s", safe, query)
+	}
 
-	logrus.WithField("duration", time.Since(startedAt)).WithField(logrus.ErrorKey, err).Debug("executed sql query ", query, " args: ", args)
+	logrus.WithField("duration", time.Since(startedAt)).WithField(logrus.ErrorKey, err).WithField("safe", safe).Debug("executed sql query ", query, " args: ", args)
 
 	return rows, err
 }
@@ -48,8 +65,21 @@ func (in *traceInterceptor) StmtQueryContext(ctx context.Context, stmt driver.St
 func (in *traceInterceptor) StmtExecContext(ctx context.Context, stmt driver.StmtExecContext, query string, args []driver.NamedValue) (driver.Result, error) {
 	startedAt := time.Now()
 	result, err := stmt.ExecContext(ctx, args)
+	key := ctx.Value(CtxDBInstance)
+	var safe string
+	if key != nil {
+		w := dbToWriter[key.(string)]
+		if w == nil {
+			safe = fmt.Sprintf("no writer for key %s", key)
+		} else {
+			safe = w.Safe()
+		}
+	}
+	if safe != "" {
+		logrus.Infof("unsafe: %s -- %s", safe, query)
+	}
 
-	logrus.WithField("duration", time.Since(startedAt)).WithField(logrus.ErrorKey, err).Debug("executed sql query ", query, " args: ", args)
+	logrus.WithField("duration", time.Since(startedAt)).WithField(logrus.ErrorKey, err).WithField("safe", safe).Debug("executed sql query ", query, " args: ", args)
 
 	return result, err
 }
@@ -75,6 +105,18 @@ func (in *traceInterceptor) RowsNext(c context.Context, rows driver.Rows, dest [
 	return err
 }
 
+func OpenWithWriter(dbProperties *config.DatabaseOptions, w Writer) (*sql.DB, context.Context, error) {
+	db, err := Open(dbProperties)
+	if err != nil {
+		return nil, nil, err
+	}
+	instCount++
+	ctxVal := fmt.Sprintf("%d", instCount)
+	dbToWriter[ctxVal] = w
+	ctx := context.WithValue(context.TODO(), CtxDBInstance, ctxVal)
+	return db, ctx, nil
+}
+
 // Open opens a database specified by its database driver name and a driver-specific data source name,
 // usually consisting of at least a database name and connection information. Includes tracing driver
 // if DENDRITE_TRACE_SQL=1
@@ -118,4 +160,5 @@ func Open(dbProperties *config.DatabaseOptions) (*sql.DB, error) {
 
 func init() {
 	registerDrivers()
+	dbToWriter = make(map[string]Writer)
 }

internal/sqlutil/writer.go

@@ -43,4 +43,6 @@ type Writer interface {
 	// Queue up one or more database write operations within the
 	// provided function to be executed when it is safe to do so.
 	Do(db *sql.DB, txn *sql.Tx, f func(txn *sql.Tx) error) error
+
+	Safe() string
 }

internal/sqlutil/writer_dummy.go

@@ -26,3 +26,7 @@ func (w *DummyWriter) Do(db *sql.DB, txn *sql.Tx, f func(txn *sql.Tx) error) err
 		return f(txn)
 	}
 }
+
+func (w *DummyWriter) Safe() string {
+	return "DummyWriter"
+}

internal/sqlutil/writer_exclusive.go

@@ -3,6 +3,10 @@ package sqlutil
 import (
 	"database/sql"
 	"errors"
+	"fmt"
+	"runtime"
+	"strconv"
+	"strings"
 
 	"go.uber.org/atomic"
 )
@@ -12,8 +16,9 @@ import (
 // contend on database locks in, e.g. SQLite. Only one task will run
 // at a time on a given ExclusiveWriter.
 type ExclusiveWriter struct {
-	running atomic.Bool
-	todo    chan transactionWriterTask
+	running  atomic.Bool
+	todo     chan transactionWriterTask
+	writerID int
 }
 
 func NewExclusiveWriter() Writer {
@@ -30,6 +35,15 @@ type transactionWriterTask struct {
 	wait chan error
 }
 
+func (w *ExclusiveWriter) Safe() string {
+	a := goid()
+	b := w.writerID
+	if a == b {
+		return ""
+	}
+	return fmt.Sprintf("%v != %v", a, b)
+}
+
 // Do queues a task to be run by a TransactionWriter. The function
 // provided will be ran within a transaction as supplied by the
 // txn parameter if one is supplied, and if not, will take out a
@@ -60,6 +74,7 @@ func (w *ExclusiveWriter) run() {
 	if !w.running.CAS(false, true) {
 		return
 	}
+	w.writerID = goid()
 	defer w.running.Store(false)
 	for task := range w.todo {
 		if task.db != nil && task.txn != nil {
@@ -74,3 +89,14 @@ func (w *ExclusiveWriter) run() {
 		close(task.wait)
 	}
 }
+
+func goid() int {
+	var buf [64]byte
+	n := runtime.Stack(buf[:], false)
+	idField := strings.Fields(strings.TrimPrefix(string(buf[:n]), "goroutine "))[0]
+	id, err := strconv.Atoi(idField)
+	if err != nil {
+		panic(fmt.Sprintf("cannot get goroutine id: %v", err))
+	}
+	return id
+}