feat(storage): read only transactions (#6417)

feat(storage): read only transactions for queries (#6415) * fix: tests * bastle wie en grosse * fix(database): scan as callback * fix tests * fix merge failures * remove as of system time * refactor: remove unused test * refacotr: remove unused lines
2025-08-12 01:47:33 +00:00 · 2023-08-22 14:49:02 +02:00
parent 7442492b8a
commit 22af4dcd97
128 changed files with 1355 additions and 897 deletions
--- a/internal/eventstore/handler/crdb/current_sequence.go
+++ b/internal/eventstore/handler/crdb/current_sequence.go
@@ -12,9 +12,10 @@ import (
 )

 const (
-	currentSequenceStmtFormat          = `SELECT current_sequence, aggregate_type, instance_id FROM %s WHERE projection_name = $1 AND instance_id = ANY ($2) FOR UPDATE`
-	updateCurrentSequencesStmtFormat   = `INSERT INTO %s (projection_name, aggregate_type, current_sequence, instance_id, timestamp) VALUES `
-	updateCurrentSequencesConflictStmt = ` ON CONFLICT (projection_name, aggregate_type, instance_id) DO UPDATE SET current_sequence = EXCLUDED.current_sequence, timestamp = EXCLUDED.timestamp`
+	currentSequenceStmtFormat            = `SELECT current_sequence, aggregate_type, instance_id FROM %s WHERE projection_name = $1 AND instance_id = ANY ($2) FOR UPDATE`
+	currentSequenceStmtWithoutLockFormat = `SELECT current_sequence, aggregate_type, instance_id FROM %s WHERE projection_name = $1 AND instance_id = ANY ($2)`
+	updateCurrentSequencesStmtFormat     = `INSERT INTO %s (projection_name, aggregate_type, current_sequence, instance_id, timestamp) VALUES `
+	updateCurrentSequencesConflictStmt   = ` ON CONFLICT (projection_name, aggregate_type, instance_id) DO UPDATE SET current_sequence = EXCLUDED.current_sequence, timestamp = EXCLUDED.timestamp`
 )

 type currentSequences map[eventstore.AggregateType][]*instanceSequence
@@ -24,41 +25,38 @@ type instanceSequence struct {
 	sequence   uint64
 }

-func (h *StatementHandler) currentSequences(ctx context.Context, query func(context.Context, string, ...interface{}) (*sql.Rows, error), instanceIDs database.StringArray) (currentSequences, error) {
-	rows, err := query(ctx, h.currentSequenceStmt, h.ProjectionName, instanceIDs)
+func (h *StatementHandler) currentSequences(ctx context.Context, isTx bool, query func(context.Context, func(*sql.Rows) error, string, ...interface{}) error, instanceIDs database.StringArray) (currentSequences, error) {
+	stmt := h.currentSequenceStmt
+	if !isTx {
+		stmt = h.currentSequenceWithoutLockStmt
+	}
+
+	sequences := make(currentSequences, len(h.aggregates))
+	err := query(ctx,
+		func(rows *sql.Rows) error {
+			for rows.Next() {
+				var (
+					aggregateType eventstore.AggregateType
+					sequence      uint64
+					instanceID    string
+				)
+
+				err := rows.Scan(&sequence, &aggregateType, &instanceID)
+				if err != nil {
+					return errors.ThrowInternal(err, "CRDB-dbatK", "scan failed")
+				}
+
+				sequences[aggregateType] = append(sequences[aggregateType], &instanceSequence{
+					sequence:   sequence,
+					instanceID: instanceID,
+				})
+			}
+			return nil
+		},
+		stmt, h.ProjectionName, instanceIDs)
 	if err != nil {
 		return nil, err
 	}
-
-	defer rows.Close()
-
-	sequences := make(currentSequences, len(h.aggregates))
-	for rows.Next() {
-		var (
-			aggregateType eventstore.AggregateType
-			sequence      uint64
-			instanceID    string
-		)
-
-		err = rows.Scan(&sequence, &aggregateType, &instanceID)
-		if err != nil {
-			return nil, errors.ThrowInternal(err, "CRDB-dbatK", "scan failed")
-		}
-
-		sequences[aggregateType] = append(sequences[aggregateType], &instanceSequence{
-			sequence:   sequence,
-			instanceID: instanceID,
-		})
-	}
-
-	if err = rows.Close(); err != nil {
-		return nil, errors.ThrowInternal(err, "CRDB-h5i5m", "close rows failed")
-	}
-
-	if err = rows.Err(); err != nil {
-		return nil, errors.ThrowInternal(err, "CRDB-O8zig", "errors in scanning rows")
-	}
-
 	return sequences, nil
 }

--- a/internal/eventstore/handler/crdb/db_mock_test.go
+++ b/internal/eventstore/handler/crdb/db_mock_test.go
@@ -124,13 +124,17 @@ func expectSavePointRelease() func(sqlmock.Sqlmock) {
 	}
 }

-func expectCurrentSequence(tableName, projection string, seq uint64, aggregateType string, instanceIDs []string) func(sqlmock.Sqlmock) {
+func expectCurrentSequence(isTx bool, tableName, projection string, seq uint64, aggregateType string, instanceIDs []string) func(sqlmock.Sqlmock) {
 	rows := sqlmock.NewRows([]string{"current_sequence", "aggregate_type", "instance_id"})
 	for _, instanceID := range instanceIDs {
 		rows.AddRow(seq, aggregateType, instanceID)
 	}
 	return func(m sqlmock.Sqlmock) {
-		m.ExpectQuery(`SELECT current_sequence, aggregate_type, instance_id FROM `+tableName+` WHERE projection_name = \$1 AND instance_id = ANY \(\$2\) FOR UPDATE`).
+		stmt := `SELECT current_sequence, aggregate_type, instance_id FROM ` + tableName + ` WHERE projection_name = \$1 AND instance_id = ANY \(\$2\)`
+		if isTx {
+			stmt += " FOR UPDATE"
+		}
+		m.ExpectQuery(stmt).
 			WithArgs(
 				projection,
 				database.StringArray(instanceIDs),
@@ -141,9 +145,13 @@ func expectCurrentSequence(tableName, projection string, seq uint64, aggregateTy
 	}
 }

-func expectCurrentSequenceErr(tableName, projection string, instanceIDs []string, err error) func(sqlmock.Sqlmock) {
+func expectCurrentSequenceErr(isTx bool, tableName, projection string, instanceIDs []string, err error) func(sqlmock.Sqlmock) {
 	return func(m sqlmock.Sqlmock) {
-		m.ExpectQuery(`SELECT current_sequence, aggregate_type, instance_id FROM `+tableName+` WHERE projection_name = \$1 AND instance_id = ANY \(\$2\) FOR UPDATE`).
+		stmt := `SELECT current_sequence, aggregate_type, instance_id FROM ` + tableName + ` WHERE projection_name = \$1 AND instance_id = ANY \(\$2\)`
+		if isTx {
+			stmt += " FOR UPDATE"
+		}
+		m.ExpectQuery(stmt).
 			WithArgs(
 				projection,
 				database.StringArray(instanceIDs),
--- a/internal/eventstore/handler/crdb/handler_stmt.go
+++ b/internal/eventstore/handler/crdb/handler_stmt.go
@@ -36,13 +36,14 @@ type StatementHandler struct {
 	*handler.ProjectionHandler
 	Locker

-	client                  *database.DB
-	sequenceTable           string
-	currentSequenceStmt     string
-	updateSequencesBaseStmt string
-	maxFailureCount         uint
-	failureCountStmt        string
-	setFailureCountStmt     string
+	client                         *database.DB
+	sequenceTable                  string
+	currentSequenceStmt            string
+	currentSequenceWithoutLockStmt string
+	updateSequencesBaseStmt        string
+	maxFailureCount                uint
+	failureCountStmt               string
+	setFailureCountStmt            string

 	aggregates  []eventstore.AggregateType
 	reduces     map[eventstore.EventType]handler.Reduce
@@ -77,20 +78,21 @@ func NewStatementHandler(
 	}

 	h := StatementHandler{
-		client:                     config.Client,
-		sequenceTable:              config.SequenceTable,
-		maxFailureCount:            config.MaxFailureCount,
-		currentSequenceStmt:        fmt.Sprintf(currentSequenceStmtFormat, config.SequenceTable),
-		updateSequencesBaseStmt:    fmt.Sprintf(updateCurrentSequencesStmtFormat, config.SequenceTable),
-		failureCountStmt:           fmt.Sprintf(failureCountStmtFormat, config.FailedEventsTable),
-		setFailureCountStmt:        fmt.Sprintf(setFailureCountStmtFormat, config.FailedEventsTable),
-		aggregates:                 aggregateTypes,
-		reduces:                    reduces,
-		bulkLimit:                  config.BulkLimit,
-		Locker:                     NewLocker(config.Client.DB, config.LockTable, config.ProjectionName),
-		initCheck:                  config.InitCheck,
-		initialized:                make(chan bool),
-		reduceScheduledPseudoEvent: reduceScheduledPseudoEvent,
+		client:                         config.Client,
+		sequenceTable:                  config.SequenceTable,
+		maxFailureCount:                config.MaxFailureCount,
+		currentSequenceStmt:            fmt.Sprintf(currentSequenceStmtFormat, config.SequenceTable),
+		currentSequenceWithoutLockStmt: fmt.Sprintf(currentSequenceStmtWithoutLockFormat, config.SequenceTable),
+		updateSequencesBaseStmt:        fmt.Sprintf(updateCurrentSequencesStmtFormat, config.SequenceTable),
+		failureCountStmt:               fmt.Sprintf(failureCountStmtFormat, config.FailedEventsTable),
+		setFailureCountStmt:            fmt.Sprintf(setFailureCountStmtFormat, config.FailedEventsTable),
+		aggregates:                     aggregateTypes,
+		reduces:                        reduces,
+		bulkLimit:                      config.BulkLimit,
+		Locker:                         NewLocker(config.Client.DB, config.LockTable, config.ProjectionName),
+		initCheck:                      config.InitCheck,
+		initialized:                    make(chan bool),
+		reduceScheduledPseudoEvent:     reduceScheduledPseudoEvent,
 	}

 	h.ProjectionHandler = handler.NewProjectionHandler(ctx, config.ProjectionHandlerConfig, h.reduce, h.Update, h.searchQuery, h.Lock, h.Unlock, h.initialized, reduceScheduledPseudoEvent)
@@ -114,7 +116,7 @@ func (h *StatementHandler) searchQuery(ctx context.Context, instanceIDs []string
 }

 func (h *StatementHandler) dbSearchQuery(ctx context.Context, instanceIDs []string) (*eventstore.SearchQueryBuilder, uint64, error) {
-	sequences, err := h.currentSequences(ctx, h.client.QueryContext, instanceIDs)
+	sequences, err := h.currentSequences(ctx, false, h.client.QueryContext, instanceIDs)
 	if err != nil {
 		return nil, 0, err
 	}
@@ -140,6 +142,26 @@ func (h *StatementHandler) dbSearchQuery(ctx context.Context, instanceIDs []stri
 	return queryBuilder, h.bulkLimit, nil
 }

+type transaction struct {
+	*sql.Tx
+}
+
+func (t *transaction) QueryContext(ctx context.Context, scan func(rows *sql.Rows) error, query string, args ...any) error {
+	rows, err := t.Tx.QueryContext(ctx, query, args...)
+	if err != nil {
+		return err
+	}
+	defer func() {
+		closeErr := rows.Close()
+		logging.OnError(closeErr).Info("rows.Close failed")
+	}()
+
+	if err = scan(rows); err != nil {
+		return err
+	}
+	return rows.Err()
+}
+
 // Update implements handler.Update
 func (h *StatementHandler) Update(ctx context.Context, stmts []*handler.Statement, reduce handler.Reduce) (index int, err error) {
 	if len(stmts) == 0 {
@@ -154,7 +176,7 @@ func (h *StatementHandler) Update(ctx context.Context, stmts []*handler.Statemen
 		return -1, errors.ThrowInternal(err, "CRDB-e89Gq", "begin failed")
 	}

-	sequences, err := h.currentSequences(ctx, tx.QueryContext, instanceIDs)
+	sequences, err := h.currentSequences(ctx, true, (&transaction{Tx: tx}).QueryContext, instanceIDs)
 	if err != nil {
 		tx.Rollback()
 		return -1, err
--- a/internal/eventstore/handler/crdb/handler_stmt_test.go
+++ b/internal/eventstore/handler/crdb/handler_stmt_test.go
@@ -90,7 +90,9 @@ func TestProjectionHandler_SearchQuery(t *testing.T) {
 					return errors.Is(err, sql.ErrTxDone)
 				},
 				expectations: []mockExpectation{
-					expectCurrentSequenceErr("my_sequences", "my_projection", []string{"instanceID1"}, sql.ErrTxDone),
+					expectBegin(),
+					expectCurrentSequenceErr(false, "my_sequences", "my_projection", []string{"instanceID1"}, sql.ErrTxDone),
+					expectRollback(),
 				},
 				SearchQueryBuilder: nil,
 			},
@@ -112,7 +114,9 @@ func TestProjectionHandler_SearchQuery(t *testing.T) {
 					return err == nil
 				},
 				expectations: []mockExpectation{
-					expectCurrentSequence("my_sequences", "my_projection", 5, "testAgg", []string{"instanceID1"}),
+					expectBegin(),
+					expectCurrentSequence(false, "my_sequences", "my_projection", 5, "testAgg", []string{"instanceID1"}),
+					expectCommit(),
 				},
 				SearchQueryBuilder: eventstore.
 					NewSearchQueryBuilder(eventstore.ColumnsEvent).
@@ -142,7 +146,9 @@ func TestProjectionHandler_SearchQuery(t *testing.T) {
 					return err == nil
 				},
 				expectations: []mockExpectation{
-					expectCurrentSequence("my_sequences", "my_projection", 5, "testAgg", []string{"instanceID1", "instanceID2"}),
+					expectBegin(),
+					expectCurrentSequence(false, "my_sequences", "my_projection", 5, "testAgg", []string{"instanceID1", "instanceID2"}),
+					expectCommit(),
 				},
 				SearchQueryBuilder: eventstore.
 					NewSearchQueryBuilder(eventstore.ColumnsEvent).
@@ -216,6 +222,7 @@ func TestProjectionHandler_SearchQuery(t *testing.T) {
 				t.Errorf("ProjectionHandler.prepareBulkStmts() error = %v", err)
 				return
 			}
+
 			if !reflect.DeepEqual(query, tt.want.SearchQueryBuilder) {
 				t.Errorf("unexpected query: expected %v, got %v", tt.want.SearchQueryBuilder, query)
 			}
@@ -289,7 +296,7 @@ func TestStatementHandler_Update(t *testing.T) {
 			want: want{
 				expectations: []mockExpectation{
 					expectBegin(),
-					expectCurrentSequenceErr("my_sequences", "my_projection", []string{"instanceID"}, sql.ErrTxDone),
+					expectCurrentSequenceErr(false, "my_sequences", "my_projection", []string{"instanceID"}, sql.ErrTxDone),
 					expectRollback(),
 				},
 				isErr: func(err error) bool {
@@ -321,7 +328,7 @@ func TestStatementHandler_Update(t *testing.T) {
 			want: want{
 				expectations: []mockExpectation{
 					expectBegin(),
-					expectCurrentSequence("my_sequences", "my_projection", 5, "testAgg", []string{"instanceID"}),
+					expectCurrentSequence(false, "my_sequences", "my_projection", 5, "testAgg", []string{"instanceID"}),
 					expectRollback(),
 				},
 				isErr: func(err error) bool {
@@ -360,7 +367,7 @@ func TestStatementHandler_Update(t *testing.T) {
 			want: want{
 				expectations: []mockExpectation{
 					expectBegin(),
-					expectCurrentSequence("my_sequences", "my_projection", 5, "testAgg", []string{"instanceID"}),
+					expectCurrentSequence(false, "my_sequences", "my_projection", 5, "testAgg", []string{"instanceID"}),
 					expectCommit(),
 				},
 				isErr: func(err error) bool {
@@ -399,7 +406,7 @@ func TestStatementHandler_Update(t *testing.T) {
 			want: want{
 				expectations: []mockExpectation{
 					expectBegin(),
-					expectCurrentSequence("my_sequences", "my_projection", 5, "agg", []string{"instanceID"}),
+					expectCurrentSequence(false, "my_sequences", "my_projection", 5, "agg", []string{"instanceID"}),
 					expectSavePoint(),
 					expectCreate("my_projection", []string{"col"}, []string{"$1"}),
 					expectSavePointRelease(),
@@ -442,7 +449,7 @@ func TestStatementHandler_Update(t *testing.T) {
 			want: want{
 				expectations: []mockExpectation{
 					expectBegin(),
-					expectCurrentSequence("my_sequences", "my_projection", 5, "agg", []string{"instanceID"}),
+					expectCurrentSequence(false, "my_sequences", "my_projection", 5, "agg", []string{"instanceID"}),
 					expectSavePoint(),
 					expectCreate("my_projection", []string{"col"}, []string{"$1"}),
 					expectSavePointRelease(),
@@ -478,7 +485,7 @@ func TestStatementHandler_Update(t *testing.T) {
 			want: want{
 				expectations: []mockExpectation{
 					expectBegin(),
-					expectCurrentSequence("my_sequences", "my_projection", 5, "testAgg", []string{"instanceID"}),
+					expectCurrentSequence(false, "my_sequences", "my_projection", 5, "testAgg", []string{"instanceID"}),
 					expectUpdateCurrentSequence("my_sequences", "my_projection", 7, "testAgg", "instanceID"),
 					expectCommit(),
 				},
@@ -511,7 +518,7 @@ func TestStatementHandler_Update(t *testing.T) {
 			want: want{
 				expectations: []mockExpectation{
 					expectBegin(),
-					expectCurrentSequence("my_sequences", "my_projection", 5, "testAgg", []string{"instanceID"}),
+					expectCurrentSequence(false, "my_sequences", "my_projection", 5, "testAgg", []string{"instanceID"}),
 					expectUpdateCurrentSequence("my_sequences", "my_projection", 7, "testAgg", "instanceID"),
 					expectCommit(),
 				},
@@ -551,7 +558,7 @@ func TestStatementHandler_Update(t *testing.T) {
 			want: want{
 				expectations: []mockExpectation{
 					expectBegin(),
-					expectCurrentSequence("my_sequences", "my_projection", 5, "testAgg", []string{"instanceID"}),
+					expectCurrentSequence(false, "my_sequences", "my_projection", 5, "testAgg", []string{"instanceID"}),
 					expectUpdateCurrentSequence("my_sequences", "my_projection", 7, "testAgg", "instanceID"),
 					expectCommit(),
 				},
@@ -1425,7 +1432,7 @@ func TestStatementHandler_currentSequence(t *testing.T) {
 					return errors.Is(err, sql.ErrConnDone)
 				},
 				expectations: []mockExpectation{
-					expectCurrentSequenceErr("my_table", "my_projection", nil, sql.ErrConnDone),
+					expectCurrentSequenceErr(true, "my_table", "my_projection", nil, sql.ErrConnDone),
 				},
 			},
 		},
@@ -1487,7 +1494,7 @@ func TestStatementHandler_currentSequence(t *testing.T) {
 					return errors.Is(err, nil)
 				},
 				expectations: []mockExpectation{
-					expectCurrentSequence("my_table", "my_projection", 5, "agg", []string{"instanceID"}),
+					expectCurrentSequence(true, "my_table", "my_projection", 5, "agg", []string{"instanceID"}),
 				},
 				sequences: currentSequences{
 					"agg": []*instanceSequence{
@@ -1515,7 +1522,7 @@ func TestStatementHandler_currentSequence(t *testing.T) {
 					return errors.Is(err, nil)
 				},
 				expectations: []mockExpectation{
-					expectCurrentSequence("my_table", "my_projection", 5, "agg", []string{"instanceID1", "instanceID2"}),
+					expectCurrentSequence(true, "my_table", "my_projection", 5, "agg", []string{"instanceID1", "instanceID2"}),
 				},
 				sequences: currentSequences{
 					"agg": []*instanceSequence{
@@ -1563,7 +1570,7 @@ func TestStatementHandler_currentSequence(t *testing.T) {
 				t.Fatalf("unexpected err in begin: %v", err)
 			}

-			seq, err := h.currentSequences(context.Background(), tx.QueryContext, tt.args.instanceIDs)
+			seq, err := h.currentSequences(context.Background(), true, (&transaction{Tx: tx}).QueryContext, tt.args.instanceIDs)
 			if !tt.want.isErr(err) {
 				t.Errorf("unexpected error: %v", err)
 			}