fix: race in TumblingWindowIncAggOp FirstTimer

Protected FirstTimer access with mutex and added thread-safe getter for testing.

Signed-off-by: Jiyong Huang <huangjy@emqx.io>

Author: ngjaying

internal/topo/node/node_test/window_inc_agg_op_test.go

@@ -261,7 +261,7 @@ func TestIncAggAlignTumblingWindow(t *testing.T) {
 	time.Sleep(10 * time.Millisecond)
 	to, ok := op.WindowExec.(*node.TumblingWindowIncAggOp)
 	require.True(t, ok)
-	require.NotNil(t, to.FirstTimer)
+	require.NotNil(t, to.GetFirstTimer())
 }
 
 func TestIncAggTumblingWindow(t *testing.T) {

internal/topo/node/window_inc_agg_op.go

@@ -386,7 +386,9 @@ func (to *TumblingWindowIncAggOp) exec(ctx api.StreamContext, errCh chan<- error
 	if !EnableAlignWindow {
 		to.ticker = timex.GetTicker(to.Interval)
 	} else {
+		to.mu.Lock()
 		_, to.FirstTimer = getFirstTimer(ctx, to.windowConfig.RawInterval, to.windowConfig.TimeUnit)
+		to.mu.Unlock()
 		if to.CurrWindow == nil {
 			to.CurrWindow = newIncAggWindow(ctx, now)
 		}
@@ -398,9 +400,9 @@ func (to *TumblingWindowIncAggOp) exec(ctx api.StreamContext, errCh chan<- error
 			case <-ctx.Done():
 				return
 			case now := <-to.FirstTimer.C:
+				to.mu.Lock()
 				to.FirstTimer.Stop()
 				to.FirstTimer = nil
-				to.mu.Lock()
 				if to.CurrWindow != nil {
 					to.emit(ctx, errCh, now)
 				}
@@ -465,6 +467,12 @@ outer:
 	}
 }
 
+func (to *TumblingWindowIncAggOp) GetFirstTimer() *clock.Timer {
+	to.mu.Lock()
+	defer to.mu.Unlock()
+	return to.FirstTimer
+}
+
 func (to *TumblingWindowIncAggOp) emit(ctx api.StreamContext, errCh chan<- error, now time.Time) {
 	results := &xsql.WindowTuples{
 		Content: make([]xsql.Row, 0),
@@ -733,7 +741,9 @@ func (ho *HoppingWindowIncAggOp) exec(ctx api.StreamContext, errCh chan<- error)
 		ho.ticker = timex.GetTicker(ho.Interval)
 		ho.newIncWindow(ctx, now)
 	} else {
+		ho.mu.Lock()
 		_, ho.FirstTimer = getFirstTimer(ctx, ho.windowConfig.RawInterval, ho.windowConfig.TimeUnit)
+		ho.mu.Unlock()
 		ho.CurrWindowList = append(ho.CurrWindowList, newIncAggWindow(ctx, now))
 	}
 	fv, _ := xsql.NewFunctionValuersForOp(ctx)
@@ -770,9 +780,9 @@ func (ho *HoppingWindowIncAggOp) exec(ctx api.StreamContext, errCh chan<- error)
 			case <-ctx.Done():
 				return
 			case now := <-ho.FirstTimer.C:
+				ho.mu.Lock()
 				ho.FirstTimer.Stop()
 				ho.FirstTimer = nil
-				ho.mu.Lock()
 				ho.CurrWindowList = gcIncAggWindow(ho.CurrWindowList, ho.Length, now)
 				ho.newIncWindow(ctx, now)
 				ho.CurrWindowList = gcIncAggWindow(ho.CurrWindowList, ho.Length, now)