matrixorigin
diff --git a/‎pkg/bootstrap/service_test.go‎
Lines changed: 1 addition & 1 deletion b/‎pkg/bootstrap/service_test.go‎
Lines changed: 1 addition & 1 deletion
diff --git a/‎pkg/frontend/test/engine_mock.go‎
Lines changed: 22 additions & 20 deletions b/‎pkg/frontend/test/engine_mock.go‎
Lines changed: 22 additions & 20 deletions
diff --git a/‎pkg/frontend/test/txn_mock.go‎
Lines changed: 1 addition & 1 deletion b/‎pkg/frontend/test/txn_mock.go‎
Lines changed: 1 addition & 1 deletion
diff --git a/‎pkg/frontend/txn_test.go‎
Lines changed: 1 addition & 1 deletion b/‎pkg/frontend/txn_test.go‎
Lines changed: 1 addition & 1 deletion
diff --git a/‎pkg/incrservice/service.go‎
Lines changed: 4 additions & 3 deletions b/‎pkg/incrservice/service.go‎
Lines changed: 4 additions & 3 deletions
diff --git a/‎pkg/incrservice/store_mem.go‎
Lines changed: 12 additions & 10 deletions b/‎pkg/incrservice/store_mem.go‎
Lines changed: 12 additions & 10 deletions
diff --git a/‎pkg/incrservice/store_sql_test.go‎
Lines changed: 1 addition & 1 deletion b/‎pkg/incrservice/store_sql_test.go‎
Lines changed: 1 addition & 1 deletion
diff --git a/‎pkg/partitionservice/storage_test.go‎
Lines changed: 23 additions & 19 deletions b/‎pkg/partitionservice/storage_test.go‎
Lines changed: 23 additions & 19 deletions
diff --git a/‎pkg/shardservice/service.go‎
Lines changed: 26 additions & 22 deletions b/‎pkg/shardservice/service.go‎
Lines changed: 26 additions & 22 deletions
diff --git a/‎pkg/shardservice/storage_test.go‎
Lines changed: 20 additions & 16 deletions b/‎pkg/shardservice/storage_test.go‎
Lines changed: 20 additions & 16 deletions
@@ -177,7 +177,7 @@ func (tTxnOp *testTxnOperator) GetWorkspace() client.Workspace {
 	panic("implement me")
 }
 
-func (tTxnOp *testTxnOperator) AppendEventCallback(event client.EventType, callbacks ...func(client.TxnEvent)) {
+func (tTxnOp *testTxnOperator) AppendEventCallback(event client.EventType, callbacks ...client.TxnEventCallback) {
 	//TODO implement me
 	panic("implement me")
 }
 
@@ -931,7 +931,7 @@ func (txnop *testTxnOp) GetWorkspace() client.Workspace {
 	return txnop.wp
 }
 
-func (txnop *testTxnOp) AppendEventCallback(event client.EventType, callbacks ...func(client.TxnEvent)) {
+func (txnop *testTxnOp) AppendEventCallback(event client.EventType, callbacks ...client.TxnEventCallback) {
 	//TODO implement me
 	panic("implement me")
 }
 
@@ -96,7 +96,7 @@ func (s *service) Create(
 
 	txnOp.AppendEventCallback(
 		client.ClosedEvent,
-		s.txnClosed)
+		client.NewTxnEventCallback(s.txnClosed))
 	if err := s.store.Create(ctx, tableID, cols, txnOp); err != nil {
 		s.logger.Error("create auto increment cache failed",
 			zap.Uint64("table-id", tableID),
@@ -180,7 +180,7 @@ func (s *service) Delete(
 
 	txnOp.AppendEventCallback(
 		client.ClosedEvent,
-		s.txnClosed)
+		client.NewTxnEventCallback(s.txnClosed))
 
 	s.mu.Lock()
 	defer s.mu.Unlock()
@@ -345,12 +345,13 @@ func (s *service) getCommittedTableCache(
 	return c, nil
 }
 
-func (s *service) txnClosed(event client.TxnEvent) {
+func (s *service) txnClosed(ctx context.Context, txnOp client.TxnOperator, event client.TxnEvent, v any) error {
 	s.mu.Lock()
 	defer s.mu.Unlock()
 
 	s.handleCreatesLocked(event.Txn)
 	s.handleDeletesLocked(event.Txn)
+	return nil
 }
 
 func (s *service) handleCreatesLocked(txnMeta txn.TxnMeta) {
 
@@ -60,17 +60,19 @@ func (s *memStore) Create(
 		s.uncommitted[string(txnOp.Txn().ID)] = m
 		txnOp.AppendEventCallback(
 			client.ClosedEvent,
-			func(event client.TxnEvent) {
-				txnMeta := event.Txn
-				s.Lock()
-				defer s.Unlock()
-				delete(s.uncommitted, string(txnMeta.ID))
-				if txnMeta.Status == txn.TxnStatus_Committed {
-					for k, v := range m {
-						s.caches[k] = v
+			client.NewTxnEventCallback(
+				func(ctx context.Context, txnOp client.TxnOperator, event client.TxnEvent, v any) error {
+					txnMeta := event.Txn
+					s.Lock()
+					defer s.Unlock()
+					delete(s.uncommitted, string(txnMeta.ID))
+					if txnMeta.Status == txn.TxnStatus_Committed {
+						for k, v := range m {
+							s.caches[k] = v
+						}
 					}
-				}
-			})
+					return nil
+				}))
 	}
 
 	caches := m[tableID]
 
@@ -285,7 +285,7 @@ func (tTxnOp *testTxnOperator) GetWorkspace() client.Workspace {
 	panic("implement me")
 }
 
-func (tTxnOp *testTxnOperator) AppendEventCallback(event client.EventType, callbacks ...func(client.TxnEvent)) {
+func (tTxnOp *testTxnOperator) AppendEventCallback(event client.EventType, callbacks ...client.TxnEventCallback) {
 	//TODO implement me
 	panic("implement me")
 }
 
@@ -116,18 +116,20 @@ func (s *memStorage) Create(
 
 	txnOp.AppendEventCallback(
 		client.ClosedEvent,
-		func(txn client.TxnEvent) {
-			s.Lock()
-			defer s.Unlock()
-
-			v, ok := s.uncommitted[def.TblId]
-			if txn.Committed() {
-				if ok {
-					s.committed[def.TblId] = v
+		client.NewTxnEventCallback(
+			func(ctx context.Context, txnOp client.TxnOperator, txn client.TxnEvent, cbdata any) error {
+				s.Lock()
+				defer s.Unlock()
+
+				v, ok := s.uncommitted[def.TblId]
+				if txn.Committed() {
+					if ok {
+						s.committed[def.TblId] = v
+					}
 				}
-			}
-			delete(s.uncommitted, def.TblId)
-		},
+				delete(s.uncommitted, def.TblId)
+				return nil
+			}),
 	)
 	return nil
 }
@@ -154,15 +156,17 @@ func (s *memStorage) Delete(
 
 	txnOp.AppendEventCallback(
 		client.ClosedEvent,
-		func(txn client.TxnEvent) {
-			s.Lock()
-			defer s.Unlock()
+		client.NewTxnEventCallback(
+			func(ctx context.Context, txnOp client.TxnOperator, txn client.TxnEvent, v any) error {
+				s.Lock()
+				defer s.Unlock()
 
-			delete(s.uncommitted, table)
-			if txn.Committed() {
-				delete(s.committed, table)
-			}
-		},
+				delete(s.uncommitted, table)
+				if txn.Committed() {
+					delete(s.committed, table)
+				}
+				return nil
+			}),
 	)
 	return nil
 }
 
@@ -181,21 +181,23 @@ func (s *service) Create(
 	if !s.options.disableAppendCreateCallback {
 		txnOp.AppendEventCallback(
 			client.ClosedEvent,
-			func(txn client.TxnEvent) {
-				if txn.Committed() {
-					s.logger.Info("sharding table created",
-						zap.Uint64("table", table),
-						zap.String("committed", txn.Txn.CommitTS.DebugString()),
-					)
-
-					// The callback here is not guaranteed to execute after the transaction has
-					// already committed.
-					// The creation will lazy execute in Read.
-					s.createC <- table
-				} else {
-					s.atomic.abort.Add(1)
-				}
-			},
+			client.NewTxnEventCallback(
+				func(ctx context.Context, _txnOp client.TxnOperator, txn client.TxnEvent, v any) error {
+					if txn.Committed() {
+						s.logger.Info("sharding table created",
+							zap.Uint64("table", table),
+							zap.String("committed", txn.Txn.CommitTS.DebugString()),
+						)
+
+						// The callback here is not guaranteed to execute after the transaction has
+						// already committed.
+						// The creation will lazy execute in Read.
+						s.createC <- table
+					} else {
+						s.atomic.abort.Add(1)
+					}
+					return nil
+				}),
 		)
 	}
 
@@ -220,13 +222,15 @@ func (s *service) Delete(
 	if !s.options.disableAppendDeleteCallback {
 		txnOp.AppendEventCallback(
 			client.ClosedEvent,
-			func(txn client.TxnEvent) {
-				if txn.Committed() {
-					s.deleteC <- table
-				} else {
-					s.atomic.abort.Add(1)
-				}
-			},
+			client.NewTxnEventCallback(
+				func(ctx context.Context, _txnOp client.TxnOperator, txn client.TxnEvent, v any) error {
+					if txn.Committed() {
+						s.deleteC <- table
+					} else {
+						s.atomic.abort.Add(1)
+					}
+					return nil
+				}),
 		)
 	}
 
 
@@ -125,15 +125,17 @@ func (s *MemShardStorage) Create(
 	}
 	txnOp.AppendEventCallback(
 		client.ClosedEvent,
-		func(txn client.TxnEvent) {
-			s.Lock()
-			defer s.Unlock()
+		client.NewTxnEventCallback(
+			func(ctx context.Context, txnOp client.TxnOperator, txn client.TxnEvent, cbdata any) error {
+				s.Lock()
+				defer s.Unlock()
 
-			if txn.Committed() {
-				s.committed[table] = v
-			}
-			delete(s.uncommittedAdd, table)
-		},
+				if txn.Committed() {
+					s.committed[table] = v
+				}
+				delete(s.uncommittedAdd, table)
+				return nil
+			}),
 	)
 	return true, nil
 }
@@ -158,15 +160,17 @@ func (s *MemShardStorage) Delete(
 	s.uncommittedDelete[table] = struct{}{}
 	txnOp.AppendEventCallback(
 		client.ClosedEvent,
-		func(txn client.TxnEvent) {
-			s.Lock()
-			defer s.Unlock()
+		client.NewTxnEventCallback(
+			func(ctx context.Context, txnOp client.TxnOperator, txn client.TxnEvent, v any) error {
+				s.Lock()
+				defer s.Unlock()
 
-			if txn.Committed() {
-				delete(s.committed, table)
-			}
-			delete(s.uncommittedDelete, table)
-		},
+				if txn.Committed() {
+					delete(s.committed, table)
+				}
+				delete(s.uncommittedDelete, table)
+				return nil
+			}),
 	)
 	return true, nil
 }
Original file line number	Diff line number	Diff line change
`@@ -177,7 +177,7 @@ func (tTxnOp *testTxnOperator) GetWorkspace() client.Workspace {`
`177`	`177`	`panic("implement me")`
`178`	`178`	`}`
`179`	`179`
`180`		`-func (tTxnOp *testTxnOperator) AppendEventCallback(event client.EventType, callbacks ...func(client.TxnEvent)) {`
	`180`	`+func (tTxnOp *testTxnOperator) AppendEventCallback(event client.EventType, callbacks ...client.TxnEventCallback) {`
`181`	`181`	`//TODO implement me`
`182`	`182`	`panic("implement me")`
`183`	`183`	`}`
Original file line number	Diff line number	Diff line change
`@@ -931,7 +931,7 @@ func (txnop *testTxnOp) GetWorkspace() client.Workspace {`
`931`	`931`	`return txnop.wp`
`932`	`932`	`}`
`933`	`933`
`934`		`-func (txnop *testTxnOp) AppendEventCallback(event client.EventType, callbacks ...func(client.TxnEvent)) {`
	`934`	`+func (txnop *testTxnOp) AppendEventCallback(event client.EventType, callbacks ...client.TxnEventCallback) {`
`935`	`935`	`//TODO implement me`
`936`	`936`	`panic("implement me")`
`937`	`937`	`}`
Original file line number	Diff line number	Diff line change
`@@ -285,7 +285,7 @@ func (tTxnOp *testTxnOperator) GetWorkspace() client.Workspace {`
`285`	`285`	`panic("implement me")`
`286`	`286`	`}`
`287`	`287`
`288`		`-func (tTxnOp *testTxnOperator) AppendEventCallback(event client.EventType, callbacks ...func(client.TxnEvent)) {`
	`288`	`+func (tTxnOp *testTxnOperator) AppendEventCallback(event client.EventType, callbacks ...client.TxnEventCallback) {`
`289`	`289`	`//TODO implement me`
`290`	`290`	`panic("implement me")`
`291`	`291`	`}`