perf(txn): avoid default event callback allocations

pkg/txn/client/client.go

@@ -191,6 +191,8 @@ type txnClient struct {
 	enableSacrificingFreshness  bool
 	enableRefreshExpression     bool
 	txnOpenedCallbacks          []func(TxnOperator)
+	defaultEventCallbacks       defaultTxnEventCallbacks
+	sharedEventCallbacks        txnEventCallbacks
 
 	// normalStateNoWait is used to control if wait for the txn client's
 	// state to be normal. If it is false, which is default value, wait
@@ -355,6 +357,11 @@ func NewTxnClient(
 		sender: sender,
 		abortC: make(chan time.Time, 1),
 	}
+	c.defaultEventCallbacks.closed = [2]TxnEventCallback{
+		{Func: c.updateLastCommitTS},
+		{Func: c.closeTxn},
+	}
+	c.sharedEventCallbacks.defaults = &c.defaultEventCallbacks
 	c.stopper = stopper.NewStopper("txn-client", stopper.WithLogger(c.logger.RawLogger()))
 	c.mu.state = paused
 	c.mu.cond = sync.NewCond(&c.mu)
@@ -462,15 +469,7 @@ func (client *txnClient) doCreateTxn(
 		op.reset.tryAcquireUnknownCommit = client.tryAcquireUnknownCommit
 		op.reset.unknownCommitResolved = client.releaseInternalUnknownCommit
 	}
-	op.AppendEventCallback(
-		ClosedEvent,
-		TxnEventCallback{
-			Func: client.updateLastCommitTS,
-		},
-		TxnEventCallback{
-			Func: client.closeTxn,
-		},
-	)
+	op.setDefaultEventCallbacks(&client.sharedEventCallbacks)
 
 	if err := client.openTxn(op); err != nil {
 		return nil, err

pkg/txn/client/operator.go

@@ -279,7 +279,7 @@ type txnOperator struct {
 		txn                    txn.TxnMeta
 		cachedWrites           map[uint64][]txn.TxnRequest
 		lockTables             []lock.LockTable
-		callbacks              map[EventType][]TxnEventCallback
+		callbacks              *txnEventCallbacks
 		retry                  bool
 		lockSeq                uint64
 		waitLocks              map[uint64]Lock
@@ -587,11 +587,7 @@ func (tc *txnOperator) initProtectedFields() {
 			delete(tc.mu.cachedWrites, k)
 		}
 	}
-	if tc.mu.callbacks != nil {
-		for k, v := range tc.mu.callbacks {
-			tc.mu.callbacks[k] = v[:0]
-		}
-	}
+	tc.mu.callbacks = nil
 	if tc.mu.waitLocks != nil {
 		for k := range tc.mu.waitLocks {
 			delete(tc.mu.waitLocks, k)

pkg/txn/client/operator_events.go

@@ -47,6 +47,28 @@ var (
 	ClosedEvent          = EventType{99, "closed"}
 )
 
+// defaultTxnEventCallbacks is initialized once by txnClient and is immutable
+// after the client is published.
+type defaultTxnEventCallbacks struct {
+	closed [2]TxnEventCallback
+}
+
+// txnEventCallbacks points at the client-owned defaults until a caller appends
+// a custom callback. The custom callback map is then allocated per operator.
+type txnEventCallbacks struct {
+	defaults  *defaultTxnEventCallbacks
+	callbacks map[EventType][]TxnEventCallback
+}
+
+func (tc *txnOperator) setDefaultEventCallbacks(callbacks *txnEventCallbacks) {
+	tc.mu.Lock()
+	defer tc.mu.Unlock()
+	if tc.mu.closed {
+		panic("set default callbacks on closed txn")
+	}
+	tc.mu.callbacks = callbacks
+}
+
 func (tc *txnOperator) AppendEventCallback(
 	event EventType,
 	callbacks ...TxnEventCallback) {
@@ -55,10 +77,17 @@ func (tc *txnOperator) AppendEventCallback(
 	if tc.mu.closed {
 		panic("append callback on closed txn")
 	}
-	if tc.mu.callbacks == nil {
-		tc.mu.callbacks = make(map[EventType][]TxnEventCallback, 1)
+	if tc.mu.callbacks == nil || tc.mu.callbacks.callbacks == nil {
+		var defaults *defaultTxnEventCallbacks
+		if tc.mu.callbacks != nil {
+			defaults = tc.mu.callbacks.defaults
+		}
+		tc.mu.callbacks = &txnEventCallbacks{
+			defaults:  defaults,
+			callbacks: make(map[EventType][]TxnEventCallback, 1),
+		}
 	}
-	tc.mu.callbacks[event] = append(tc.mu.callbacks[event], callbacks...)
+	tc.mu.callbacks.callbacks[event] = append(tc.mu.callbacks.callbacks[event], callbacks...)
 }
 
 func (tc *txnOperator) triggerEvent(ctx context.Context, event TxnEvent) error {
@@ -71,7 +100,15 @@ func (tc *txnOperator) triggerEventLocked(ctx context.Context, event TxnEvent) (
 	if tc.mu.callbacks == nil {
 		return
 	}
-	for _, cb := range tc.mu.callbacks[event.Event] {
+	if event.Event == ClosedEvent && tc.mu.callbacks.defaults != nil {
+		for _, cb := range tc.mu.callbacks.defaults.closed {
+			err = cb.Func(ctx, tc, event, cb.Value)
+			if err != nil {
+				return
+			}
+		}
+	}
+	for _, cb := range tc.mu.callbacks.callbacks[event.Event] {
 		err = cb.Func(ctx, tc, event, cb.Value)
 		if err != nil {
 			return

pkg/txn/client/operator_events_benchmark_test.go

@@ -0,0 +1,151 @@
+// Copyright 2026 Matrix Origin
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//      http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package client
+
+import (
+	"context"
+	"testing"
+
+	commonruntime "github.com/matrixorigin/matrixone/pkg/common/runtime"
+	"github.com/matrixorigin/matrixone/pkg/logutil"
+	"github.com/matrixorigin/matrixone/pkg/pb/metadata"
+	"github.com/matrixorigin/matrixone/pkg/pb/timestamp"
+	"github.com/matrixorigin/matrixone/pkg/txn/clock"
+	"go.uber.org/zap/zapcore"
+)
+
+func BenchmarkTxnNewRollback(b *testing.B) {
+	b.Run("serial", func(b *testing.B) {
+		c := newBenchmarkTxnClient(b)
+		ctx := context.Background()
+		b.ReportAllocs()
+		b.ResetTimer()
+		for i := 0; i < b.N; i++ {
+			op, err := c.New(ctx, timestamp.Timestamp{})
+			if err != nil {
+				b.Fatal(err)
+			}
+			if err := op.Rollback(ctx); err != nil {
+				b.Fatal(err)
+			}
+		}
+	})
+
+	b.Run("parallel", func(b *testing.B) {
+		c := newBenchmarkTxnClient(b)
+		ctx := context.Background()
+		b.ReportAllocs()
+		b.ResetTimer()
+		b.RunParallel(func(pb *testing.PB) {
+			for pb.Next() {
+				op, err := c.New(ctx, timestamp.Timestamp{})
+				if err != nil {
+					b.Fatal(err)
+				}
+				if err := op.Rollback(ctx); err != nil {
+					b.Fatal(err)
+				}
+			}
+		})
+	})
+}
+
+func BenchmarkTxnNewAppendCallbackRollback(b *testing.B) {
+	callback := NewTxnEventCallback(func(context.Context, TxnOperator, TxnEvent, any) error {
+		return nil
+	})
+	b.Run("serial", func(b *testing.B) {
+		c := newBenchmarkTxnClient(b)
+		ctx := context.Background()
+		b.ReportAllocs()
+		b.ResetTimer()
+		for i := 0; i < b.N; i++ {
+			op, err := c.New(ctx, timestamp.Timestamp{})
+			if err != nil {
+				b.Fatal(err)
+			}
+			op.AppendEventCallback(ClosedEvent, callback)
+			if err := op.Rollback(ctx); err != nil {
+				b.Fatal(err)
+			}
+		}
+	})
+
+	b.Run("parallel", func(b *testing.B) {
+		c := newBenchmarkTxnClient(b)
+		ctx := context.Background()
+		b.ReportAllocs()
+		b.ResetTimer()
+		b.RunParallel(func(pb *testing.PB) {
+			for pb.Next() {
+				op, err := c.New(ctx, timestamp.Timestamp{})
+				if err != nil {
+					b.Fatal(err)
+				}
+				op.AppendEventCallback(ClosedEvent, callback)
+				if err := op.Rollback(ctx); err != nil {
+					b.Fatal(err)
+				}
+			}
+		})
+	})
+}
+
+func BenchmarkTxnNewAppendTraceCallbacksRollback(b *testing.B) {
+	events := []EventType{
+		WaitActiveEvent, UpdateSnapshotEvent, CommitEvent, RollbackEvent,
+		CommitResponseEvent, CommitWaitApplyEvent, UnlockEvent, RangesEvent,
+		BuildPlanEvent, ExecuteSQLEvent, CompileEvent, TableScanEvent,
+		WorkspaceWriteEvent, WorkspaceAdjustEvent,
+	}
+	callback := NewTxnEventCallback(func(context.Context, TxnOperator, TxnEvent, any) error {
+		return nil
+	})
+	c := newBenchmarkTxnClient(b)
+	ctx := context.Background()
+	b.ReportAllocs()
+	b.ResetTimer()
+	for i := 0; i < b.N; i++ {
+		op, err := c.New(ctx, timestamp.Timestamp{})
+		if err != nil {
+			b.Fatal(err)
+		}
+		for _, event := range events {
+			op.AppendEventCallback(event, callback)
+		}
+		if err := op.Rollback(ctx); err != nil {
+			b.Fatal(err)
+		}
+	}
+}
+
+func newBenchmarkTxnClient(b *testing.B) TxnClient {
+	b.Helper()
+	rt := commonruntime.NewRuntime(
+		metadata.ServiceType_CN,
+		"",
+		logutil.GetPanicLoggerWithLevel(zapcore.ErrorLevel),
+		commonruntime.WithClock(clock.NewHLCClock(func() int64 { return 1 }, 0)),
+	)
+	commonruntime.SetupServiceBasedRuntime("", rt)
+	c := NewTxnClient("", newTestTxnSender())
+	c.Resume()
+	b.Cleanup(func() {
+		if err := c.Close(); err != nil {
+			b.Fatal(err)
+		}
+	})
+	return c
+}