chore(rule): remove rule auto restart

Will simplify it and implement in another way. Do it in the rule patrol.

Signed-off-by: Jiyong Huang <[email protected]>

Author: ngjaying

internal/topo/rule/action.go

@@ -1,13 +1,8 @@
 package rule
 
 import (
-	"context"
 	"errors"
 	"fmt"
-	"math"
-	"math/rand/v2"
-	"strings"
-	"time"
 
 	"github.com/lf-edge/ekuiper/v2/internal/pkg/def"
 	"github.com/lf-edge/ekuiper/v2/internal/topo"
@@ -100,18 +95,13 @@ func (s *State) doStart() error {
 				s.topoGraph = s.topology.GetTopo()
 			}
 		}
-		ctx, cancel := context.WithCancel(context.Background())
-		s.cancelRetry = cancel
-		go s.runTopo(ctx, s.topology, s.Rule.Options.RestartStrategy)
+		go s.runTopo(s.topology)
 		return nil
 	})
 	return err
 }
 
 func (s *State) doStop() error {
-	if s.cancelRetry != nil {
-		s.cancelRetry()
-	}
 	if s.topology != nil {
 		e := s.topology.GetContext().Err()
 		s.topoGraph = s.topology.GetTopo()
@@ -144,91 +134,45 @@ func (s *State) stopOld() {
 	return
 }
 
-// This is called async
-func (s *State) runTopo(ctx context.Context, tp *topo.Topo, rs *def.RestartStrategy) {
-	err := infra.SafeRun(func() error {
-		count := 0
-		d := time.Duration(rs.Delay)
-		var er error
-		ticker := time.NewTicker(d)
-		defer ticker.Stop()
-		for {
-			select {
-			case e := <-tp.Open():
-				er = e
-				if errorx.IsUnexpectedErr(er) { // Only restart Rule for errors
-					tp.GetContext().SetError(er)
-					s.logger.Errorf("closing Rule for error: %v", er)
-					tp.Cancel()
-					s.transitState(machine.Stopped, "retrying after error: "+er.Error())
-				} else {
-					// exit normally
-					lastWill := "cancelled manually"
-					if errorx.IsEOF(er) {
-						lastWill = EOFMessage
-						msg := er.Error()
-						if len(msg) > 0 {
-							lastWill = fmt.Sprintf("%s: %s", lastWill, msg)
-						}
-						s.updateTrigger(s.Rule.Id, false)
-					}
-					tp.Cancel()
-					s.transitState(machine.Stopped, lastWill)
-					return nil
-				}
-			}
-			if count < rs.Attempts {
-				if d > time.Duration(rs.MaxDelay) {
-					d = time.Duration(rs.MaxDelay)
-				}
-				if rs.JitterFactor > 0 {
-					d = time.Duration(math.Round(float64(d.Milliseconds())*((rand.Float64()*2-1)*rs.JitterFactor+1))) * time.Millisecond
-					// make sure d is always in range
-					for d <= 0 || d > time.Duration(rs.MaxDelay) {
-						d = time.Duration(math.Round(float64(d.Milliseconds())*((rand.Float64()*2-1)*rs.JitterFactor+1))) * time.Millisecond
-					}
-					s.logger.Infof("Rule will restart with jitterred delay %d", d)
-				} else {
-					s.logger.Infof("Rule will restart with delay %d", d)
-				}
-				// retry after delay
-				select {
-				case <-ticker.C:
-					break
-				case <-ctx.Done():
-					s.logger.Errorf("stop Rule retry as cancelled")
-					return nil
-				}
-				count++
-				if rs.Multiplier > 0 {
-					d = time.Duration(rs.Delay) * time.Duration(math.Pow(rs.Multiplier, float64(count)))
-				}
-			} else {
-				return er
+// This is called async, so do not touch state properties
+func (s *State) runTopo(tp *topo.Topo) {
+	e := <-tp.Open()
+	tp.Cancel()
+	var lastWill string
+	hasError := false
+	if errorx.IsUnexpectedErr(e) { // Only restart Rule for errors
+		tp.GetContext().SetError(e)
+		lastWill = e.Error()
+		hasError = true
+	} else {
+		// exit normally
+		lastWill = "canceled manually"
+		if errorx.IsEOF(e) {
+			lastWill = EOFMessage
+			msg := e.Error()
+			if len(msg) > 0 {
+				lastWill = fmt.Sprintf("%s: %s", lastWill, msg)
 			}
+			s.updateTrigger(s.Rule.Id, false)
 		}
-	})
-	s.cleanRule(err)
+	}
+	s.cleanRule(hasError, lastWill)
 }
 
-func (s *State) cleanRule(err error) {
+func (s *State) cleanRule(hasError bool, lastWill string) {
 	s.ruleLock.Lock()
 	defer s.ruleLock.Unlock()
 	if s.topology != nil {
 		s.topoGraph = s.topology.GetTopo()
 		keys, values := s.topology.GetMetrics()
 		s.stoppedMetrics = []any{keys, values}
 	}
-	if err != nil { // Exit after retries
-		s.logger.Error(err)
-		s.transitState(machine.StoppedByErr, err.Error())
-		s.topology = nil
+	s.topology = nil
+	if hasError {
+		s.transitState(machine.StoppedByErr, lastWill)
 		s.logger.Infof("%s exit by error set tp to nil", s.Rule.Id)
-	} else if strings.HasPrefix(s.sm.LastWill(), EOFMessage) {
-		// Two case when err is nil; 1. Manually stop 2.EOF
-		// Only transit status when EOF. Don't do this for manual stop because the state already changed!
-		s.transitState(machine.Stopped, "")
-		s.topology = nil
+	} else {
+		s.transitState(machine.Stopped, lastWill)
 		s.logger.Infof("%s exit eof set tp to nil", s.Rule.Id)
 	}
 }

internal/topo/rule/state.go

@@ -16,12 +16,12 @@ package rule
 
 import (
 	"bytes"
-	"context"
 	"encoding/json"
 	"fmt"
 	"strconv"
 	"strings"
 	"sync"
+	"sync/atomic"
 	"time"
 
 	"github.com/lf-edge/ekuiper/contract/v2/api"
@@ -50,27 +50,31 @@ type State struct {
 	logger        api.Logger
 	updateTrigger func(string, bool)
 	// The physical rule instance for each **run**. control the lifecycle in State.
-	topology    *topo.Topo
-	cancelRetry context.CancelFunc
+	topology *topo.Topo
 	// temporary storage for topo graph to make sure even Rule close, the graph is still available
 	topoGraph *def.PrintableTopo
 	// Metric RunState
 	stoppedMetrics []any
 	// State machine
-	sm machine.StateMachine
+	sm      machine.StateMachine
+	actionq chan struct{}
+	deleted atomic.Bool
 }
 
 // NewState provision a state instance only.
 // Do not plan or run as before. If the Rule is not triggered, do not plan or run.
 // When called by recover Rule, expect
 func NewState(rule *def.Rule, updateTriggerFunc func(string, bool)) *State {
 	contextLogger := conf.Log.WithField("Rule", rule.Id)
-	return &State{
+	s := &State{
 		Rule:          rule,
 		sm:            machine.NewStateMachine(contextLogger),
 		logger:        contextLogger,
 		updateTrigger: updateTriggerFunc,
+		actionq:       make(chan struct{}, 100),
 	}
+	go s.processActions()
+	return s
 }
 
 // ValidateAndRun tries to set up the rule in an atomic way
@@ -99,7 +103,6 @@ func (s *State) Bootstrap() error {
 // By check state, it assures only one Start function is running at any time. (thread safe)
 // regSchedule: whether need to handle scheduler. If call externally, set it to true
 func (s *State) Start() error {
-	s.logger.Debug("start RunState")
 	done := s.sm.TriggerAction(machine.ActionSignalStart)
 	if done {
 		return nil
@@ -209,6 +212,7 @@ func (s *State) Delete() (err error) {
 		s.topology.Cancel()
 		s.topology.WaitClose()
 	}
+	s.triggerAction(true)
 	return nil
 }
 

internal/topo/rule/state_test.go

@@ -103,7 +103,12 @@ func TestAPIs(t *testing.T) {
 	assert.NoError(t, e)
 	e = st.Start()
 	assert.NoError(t, e)
-	time.Sleep(100 * time.Millisecond)
+	for i := 0; i < 10; i++ {
+		time.Sleep(100 * time.Millisecond)
+		if st.sm.CurrentState() == machine.Running {
+			break
+		}
+	}
 	assert.Equal(t, machine.Running, st.sm.CurrentState())
 	err = st.ResetStreamOffset("test", nil)
 	assert.EqualError(t, err, "stream test not found in topo")

internal/topo/rule/transit.go

@@ -2,11 +2,28 @@ package rule
 
 import "github.com/lf-edge/ekuiper/v2/internal/topo/rule/machine"
 
+func (s *State) processActions() {
+	for range s.actionq {
+		s.nextAction()
+	}
+}
+
+func (s *State) triggerAction(stop bool) {
+	if stop {
+		s.deleted.Store(true)
+		close(s.actionq)
+		return
+	}
+	if !s.deleted.Load() {
+		s.actionq <- struct{}{}
+	}
+}
+
 func (s *State) transitState(newState machine.RunState, lastWill string) {
 	chainAction := false
 	defer func() {
 		if chainAction {
-			s.nextAction()
+			s.triggerAction(false)
 		}
 	}()
 	chainAction = s.sm.Transit(newState, lastWill)