Repository: lf-edge/ekuiper
Pull request #3083 (2024-08-07 09:24:35)

feat(state): clearer state transit

codecov[bot]: ## [Codecov](https://app.codecov.io/gh/lf-edge/ekuiper/pull/3083?dropdown=coverage&src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=lf-edge) Report
Attention: Patch coverage is `53.41365%` with `348 lines` in your changes missing coverage. Please review.
> Project coverage is 67.56%. Comparing base [(`d6457d0`)](https://app.codecov.io/gh/lf-edge/ekuiper/commit/d6457d008e129b1cdd54d76b5993992c349d1b80?dropdown=coverage&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=lf-edge) to head [(`8a2136d`)](https://app.codecov.io/gh/lf-edge/ekuiper/commit/8a2136dc6fb3579900a8ed7ce9999f71aecc9136?dropdown=coverage&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=lf-edge).

| [Files](https://app.codecov.io/gh/lf-edge/ekuiper/pull/3083?dropdown=coverage&src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=lf-edge) | Patch % | Lines |
|---|---|---|
| [internal/topo/rule/ruleState.go](https://app.codecov.io/gh/lf-edge/ekuiper/pull/3083?src=pr&el=tree&filepath=internal%2Ftopo%2Frule%2FruleState.go&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=lf-edge#diff-aW50ZXJuYWwvdG9wby9ydWxlL3J1bGVTdGF0ZS5nbw==) | 55.39% | [138 Missing and 15 partials :warning: ](https://app.codecov.io/gh/lf-edge/ekuiper/pull/3083?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=lf-edge) |
| [internal/topo/rule/cron.go](https://app.codecov.io/gh/lf-edge/ekuiper/pull/3083?src=pr&el=tree&filepath=internal%2Ftopo%2Frule%2Fcron.go&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=lf-edge#diff-aW50ZXJuYWwvdG9wby9ydWxlL2Nyb24uZ28=) | 6.45% | [83 Missing and 4 partials :warning: ](https://app.codecov.io/gh/lf-edge/ekuiper/pull/3083?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=lf-edge) |
| [internal/server/rule\_manager.go](https://app.codecov.io/gh/lf-edge/ekuiper/pull/3083?src=pr&el=tree&filepath=internal%2Fserver%2Frule_manager.go&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=lf-edge#diff-aW50ZXJuYWwvc2VydmVyL3J1bGVfbWFuYWdlci5nbw==) | 64.76% | [60 Missing and 20 partials :warning: ](https://app.codecov.io/gh/lf-edge/ekuiper/pull/3083?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=lf-edge) |
| [internal/server/server.go](https://app.codecov.io/gh/lf-edge/ekuiper/pull/3083?src=pr&el=tree&filepath=internal%2Fserver%2Fserver.go&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=lf-edge#diff-aW50ZXJuYWwvc2VydmVyL3NlcnZlci5nbw==) | 45.00% | [10 Missing and 1 partial :warning: ](https://app.codecov.io/gh/lf-edge/ekuiper/pull/3083?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=lf-edge) |
| [internal/server/rpc.go](https://app.codecov.io/gh/lf-edge/ekuiper/pull/3083?src=pr&el=tree&filepath=internal%2Fserver%2Frpc.go&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=lf-edge#diff-aW50ZXJuYWwvc2VydmVyL3JwYy5nbw==) | 61.54% | [3 Missing and 7 partials :warning: ](https://app.codecov.io/gh/lf-edge/ekuiper/pull/3083?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=lf-edge) |
| [internal/server/rule\_state.go](https://app.codecov.io/gh/lf-edge/ekuiper/pull/3083?src=pr&el=tree&filepath=internal%2Fserver%2Frule_state.go&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=lf-edge#diff-aW50ZXJuYWwvc2VydmVyL3J1bGVfc3RhdGUuZ28=) | 40.00% | [3 Missing :warning: ](https://app.codecov.io/gh/lf-edge/ekuiper/pull/3083?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=lf-edge) |
| [internal/processor/rule.go](https://app.codecov.io/gh/lf-edge/ekuiper/pull/3083?src=pr&el=tree&filepath=internal%2Fprocessor%2Frule.go&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=lf-edge#diff-aW50ZXJuYWwvcHJvY2Vzc29yL3J1bGUuZ28=) | 77.78% | [2 Missing :warning: ](https://app.codecov.io/gh/lf-edge/ekuiper/pull/3083?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=lf-edge) |
| [internal/server/import\_export.go](https://app.codecov.io/gh/lf-edge/ekuiper/pull/3083?src=pr&el=tree&filepath=internal%2Fserver%2Fimport_export.go&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=lf-edge#diff-aW50ZXJuYWwvc2VydmVyL2ltcG9ydF9leHBvcnQuZ28=) | 33.33% | [2 Missing :warning: ](https://app.codecov.io/gh/lf-edge/ekuiper/pull/3083?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=lf-edge) |

<details><summary>Additional details and impacted files</summary>


```diff
@@            Coverage Diff             @@
##           master    #3083      +/-   ##
==========================================
- Coverage   68.09%   67.56%   -0.53%     
==========================================
  Files         390      391       +1     
  Lines       37121    37146      +25     
==========================================
- Hits        25275    25096     -179     
- Misses       9570     9781     +211     
+ Partials     2276     2269       -7     
```

</details>

[:umbrella: View full report in Codecov by Sentry](https://app.codecov.io/gh/lf-edge/ekuiper/pull/3083?dropdown=coverage&src=pr&el=continue&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=lf-edge).   
:loudspeaker: Have feedback on the report? [Share it here](https://about.codecov.io/codecov-pr-comment-feedback/?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=lf-edge).


Patch:
diff --git a/internal/processor/rule.go b/internal/processor/rule.go
index 2668f881..8fd55435 100644
--- a/internal/processor/rule.go
+++ b/internal/processor/rule.go
@@ -1,4 +1,4 @@
-// Copyright 2021-2023 EMQ Technologies Co., Ltd.
+// Copyright 2021-2024 EMQ Technologies Co., Ltd.
 //
 // Licensed under the Apache License, Version 2.0 (the "License");
 // you may not use this file except in compliance with the License.
@@ -17,6 +17,7 @@ package processor
 import (
 	"bytes"
 	"encoding/json"
+	"errors"
 	"fmt"
 
 	"github.com/lf-edge/ekuiper/v2/internal/conf"
@@ -251,24 +252,25 @@ func (p *RuleProcessor) GetAllRulesJson() (map[string]string, error) {
 	return p.db.All()
 }
 
-func (p *RuleProcessor) ExecDrop(name string) (string, error) {
-	result := fmt.Sprintf("Rule %s is dropped.", name)
-	var ruleJson string
+func (p *RuleProcessor) ExecDrop(name string) error {
+	var (
+		ruleJson string
+		allErr   error
+	)
 	if ok, _ := p.db.Get(name, &ruleJson); ok {
 		if err := cleanSinkCache(name); err != nil {
-			result = fmt.Sprintf("%s. Clean sink cache faile: %s.", result, err)
+			allErr = errors.Join(allErr, fmt.Errorf("Clean sink cache failed: %v.", err))
 		}
 		if err := cleanCheckpoint(name); err != nil {
-			result = fmt.Sprintf("%s. Clean checkpoint cache faile: %s.", result, err)
+			allErr = errors.Join(allErr, fmt.Errorf("Clean checkpoint cache failed: %v.", err))
 		}
 
 	}
 	err := p.db.Delete(name)
 	if err != nil {
-		return "", err
-	} else {
-		return result, nil
+		allErr = errors.Join(allErr, fmt.Errorf("Delete rule %s failed: %v.", name, err))
 	}
+	return allErr
 }
 
 func cleanCheckpoint(name string) error {
diff --git a/internal/server/import_export.go b/internal/server/import_export.go
index f0be49bf..c898951b 100644
--- a/internal/server/import_export.go
+++ b/internal/server/import_export.go
@@ -259,7 +259,7 @@ func configurationImport(ctx context.Context, data []byte, reboot bool) ImportCo
 					logger.Error(ee)
 					continue
 				}
-				reply := recoverRule(rul)
+				reply := registry.RecoverRule(rul)
 				if reply != "" {
 					logger.Error(reply)
 				}
@@ -585,20 +585,14 @@ func importRuleSetPartial(all processor.Ruleset) processor.Ruleset {
 		_, err := ruleProcessor.GetRuleJson(k)
 		if err == nil {
 			// the rule already exist, update
-			err = updateRule(k, v, false)
-			if err != nil {
-				ruleSetRsp.Rules[k] = err.Error()
-				continue
-			}
-			// Update to db after validation
-			_, err = ruleProcessor.ExecUpdate(k, v)
+			err = registry.UpdateRule(k, v)
 			if err != nil {
 				ruleSetRsp.Rules[k] = err.Error()
 				continue
 			}
 		} else {
 			// not found, create
-			_, err2 := createRule(k, v)
+			_, err2 := registry.CreateRule(k, v)
 			if err2 != nil {
 				ruleSetRsp.Rules[k] = err2.Error()
 				continue
diff --git a/internal/server/rest.go b/internal/server/rest.go
index 6b4f805d..1b90f128 100644
--- a/internal/server/rest.go
+++ b/internal/server/rest.go
@@ -513,11 +513,11 @@ func checkStreamBeforeDrop(name string) (bool, error) {
 		return false, err
 	}
 	for _, r := range rules {
-		rs, ok := registry.Load(r)
+		rs, ok := registry.load(r)
 		if !ok {
 			continue
 		}
-		streams := rs.Topology.GetStreams()
+		streams := rs.GetStreams()
 		for _, s := range streams {
 			if name == s {
 				return true, nil
@@ -635,7 +635,7 @@ func rulesHandler(w http.ResponseWriter, r *http.Request) {
 			handleError(w, err, "Invalid body", logger)
 			return
 		}
-		id, err := createRule("", string(body))
+		id, err := registry.CreateRule("", string(body))
 		if err != nil {
 			handleError(w, err, "", logger)
 			return
@@ -643,7 +643,7 @@ func rulesHandler(w http.ResponseWriter, r *http.Request) {
 		w.WriteHeader(http.StatusCreated)
 		fmt.Fprintf(w, "Rule %s was created successfully.", id)
 	case http.MethodGet:
-		content, err := getAllRulesWithStatus()
+		content, err := registry.GetAllRulesWithStatus()
 		if err != nil {
 			handleError(w, err, "Show rules error", logger)
 			return
@@ -668,17 +668,15 @@ func ruleHandler(w http.ResponseWriter, r *http.Request) {
 		w.Header().Add(ContentType, ContentTypeJSON)
 		w.Write([]byte(rule))
 	case http.MethodDelete:
-		deleteRule(name)
-		content, err := ruleProcessor.ExecDrop(name)
+		// delete rule will wait until rule close
+		err := registry.DeleteRule(name)
 		if err != nil {
 			handleError(w, err, "Delete rule error", logger)
 			return
 		}
-		// wait resource clear
-		time.Sleep(time.Second)
 		conf.Log.Infof("drop rule:%v", name)
 		w.WriteHeader(http.StatusOK)
-		w.Write([]byte(content))
+		_, _ = fmt.Fprintf(w, "Rule %s is dropped.", name)
 	case http.MethodPut:
 		_, err := ruleProcessor.GetRuleById(name)
 		if err != nil {
@@ -691,30 +689,19 @@ func ruleHandler(w http.ResponseWriter, r *http.Request) {
 			handleError(w, err, "Invalid body", logger)
 			return
 		}
-		newRuleJson, err := replaceRulePassword(name, string(body))
-		if err != nil {
-			handleError(w, err, "Invalid body", logger)
-			return
-		}
-		err = updateRule(name, newRuleJson, true)
+		err = registry.UpdateRule(name, string(body))
 		if err != nil {
 			handleError(w, err, "Update rule error", logger)
 			return
 		}
-		// Update to db after validation
-		_, err = ruleProcessor.ExecUpdate(name, newRuleJson)
-		if err != nil {
-			handleError(w, err, "Update rule error, suggest to delete it and recreate", logger)
-			return
-		}
 		w.WriteHeader(http.StatusOK)
-		fmt.Fprintf(w, "Rule %s was updated successfully.", name)
+		_, _ = fmt.Fprintf(w, "Rule %s was updated successfully.", name)
 	}
 }
 
 func getAllRuleStatusHandler(w http.ResponseWriter, r *http.Request) {
 	defer r.Body.Close()
-	s, err := getAllRuleStatus()
+	s, err := registry.GetAllRuleStatus()
 	if err != nil {
 		handleError(w, err, "get rules status error", logger)
 		return
@@ -730,7 +717,7 @@ func getStatusV2RulHandler(w http.ResponseWriter, r *http.Request) {
 	vars := mux.Vars(r)
 	name := vars["name"]
 
-	content, err := getRuleStatusV2(name)
+	content, err := registry.GetRuleStatusV2(name)
 	if err != nil {
 		handleError(w, err, "get rule status error", logger)
 		return
@@ -745,7 +732,7 @@ func getStatusRuleHandler(w http.ResponseWriter, r *http.Request) {
 	vars := mux.Vars(r)
 	name := vars["name"]
 
-	content, err := getRuleStatus(name)
+	content, err := registry.GetRuleStatus(name)
 	if err != nil {
 		handleError(w, err, "get rule status error", logger)
 		return
@@ -761,13 +748,13 @@ func startRuleHandler(w http.ResponseWriter, r *http.Request) {
 	vars := mux.Vars(r)
 	name := vars["name"]
 
-	err := startRule(name)
+	err := registry.StartRule(name)
 	if err != nil {
 		handleError(w, err, "start rule error", logger)
 		return
 	}
 	w.WriteHeader(http.StatusOK)
-	fmt.Fprintf(w, "Rule %s was started", name)
+	_, _ = fmt.Fprintf(w, "Rule %s was started", name)
 }
 
 // stop a rule
@@ -776,15 +763,13 @@ func stopRuleHandler(w http.ResponseWriter, r *http.Request) {
 	vars := mux.Vars(r)
 	name := vars["name"]
 
-	result, err := stopRule(name)
+	err := registry.StopRule(name)
 	if err != nil {
 		handleError(w, err, "stop rule error", logger)
 		return
 	}
-	// wait resource clear
-	time.Sleep(time.Second)
 	w.WriteHeader(http.StatusOK)
-	w.Write([]byte(result))
+	_, _ = fmt.Fprintf(w, "Rule %s was stopped.", name)
 }
 
 // restart a rule
@@ -793,7 +778,7 @@ func restartRuleHandler(w http.ResponseWriter, r *http.Request) {
 	vars := mux.Vars(r)
 	name := vars["name"]
 
-	err := restartRule(name)
+	err := registry.RestartRule(name)
 	if err != nil {
 		handleError(w, err, "restart rule error", logger)
 		return
@@ -808,7 +793,7 @@ func getTopoRuleHandler(w http.ResponseWriter, r *http.Request) {
 	vars := mux.Vars(r)
 	name := vars["name"]
 
-	content, err := getRuleTopo(name)
+	content, err := registry.GetRuleTopo(name)
 	if err != nil {
 		handleError(w, err, "get rule topo error", logger)
 		return
@@ -825,7 +810,7 @@ func validateRuleHandler(w http.ResponseWriter, r *http.Request) {
 		handleError(w, err, "Invalid body", logger)
 		return
 	}
-	sources, validate, err := validateRule("", string(body))
+	sources, validate, err := registry.ValidateRule("", string(body))
 	if !validate {
 		w.WriteHeader(http.StatusUnprocessableEntity)
 		w.Write([]byte(err.Error()))
@@ -886,7 +871,7 @@ func importHandler(w http.ResponseWriter, r *http.Request) {
 				logger.Error(ee)
 				continue
 			}
-			reply := recoverRule(rul)
+			reply := registry.RecoverRule(rul)
 			if reply != "" {
 				logger.Error(reply)
 			}
@@ -947,5 +932,5 @@ func testRuleStopHandler(w http.ResponseWriter, r *http.Request) {
 	id := vars["name"]
 	trial.TrialManager.StopRule(id)
 	w.WriteHeader(http.StatusOK)
-	fmt.Fprintf(w, "Test rule %s was stopped", id)
+	fmt.Fprintf(w, "Test rule %s was stopped.", id)
 }
diff --git a/internal/server/rpc.go b/internal/server/rpc.go
index 84f7eef5..3528a0d6 100644
--- a/internal/server/rpc.go
+++ b/internal/server/rpc.go
@@ -30,6 +30,7 @@ import (
 
 	"github.com/lf-edge/ekuiper/v2/internal/conf"
 	"github.com/lf-edge/ekuiper/v2/internal/io/sink"
+	"github.com/lf-edge/ekuiper/v2/internal/pkg/def"
 	"github.com/lf-edge/ekuiper/v2/internal/pkg/model"
 	"github.com/lf-edge/ekuiper/v2/internal/topo/rule"
 	"github.com/lf-edge/ekuiper/v2/pkg/cast"
@@ -86,15 +87,17 @@ func (r *rpcComp) close() {
 type Server int
 
 func (t *Server) CreateQuery(sql string, reply *string) error {
-	if _, ok := registry.Load(QueryRuleId); ok {
+	if _, ok := registry.load(QueryRuleId); ok {
 		stopQuery()
 	}
 	tp, err := ruleProcessor.ExecQuery(QueryRuleId, sql)
 	if err != nil {
 		return err
 	} else {
-		rs := &rule.RuleState{RuleId: QueryRuleId, Topology: tp}
-		registry.Store(QueryRuleId, rs)
+		rs := rule.NewState(def.GetDefaultRule(QueryRuleId, sql))
+		rs.WithTopo(tp)
+		registry.register(QueryRuleId, rs)
+		_ = rs.Start()
 		msg := fmt.Sprintf("Query was submit successfully.")
 		logger.Println(msg)
 		*reply = fmt.Sprint(msg)
@@ -103,10 +106,9 @@ func (t *Server) CreateQuery(sql string, reply *string) error {
 }
 
 func stopQuery() {
-	if rs, ok := registry.Load(QueryRuleId); ok {
+	if rs, ok := registry.load(QueryRuleId); ok {
 		logger.Printf("stop the query.")
-		(*rs.Topology).Cancel()
-		registry.Delete(QueryRuleId)
+		_ = rs.Delete()
 	}
 }
 
@@ -114,10 +116,10 @@ func stopQuery() {
  * qid is not currently used.
  */
 func (t *Server) GetQueryResult(_ string, reply *string) error {
-	if rs, ok := registry.Load(QueryRuleId); ok {
-		c := (*rs.Topology).GetContext()
-		if c != nil && c.Err() != nil {
-			return c.Err()
+	if rs, ok := registry.load(QueryRuleId); ok {
+		st := rs.GetState()
+		if st == rule.Stopped || st == rule.StoppedByErr {
+			return fmt.Errorf("query rule is stopped: %s", rs.GetLastWill())
 		}
 	}
 
@@ -146,7 +148,7 @@ func (t *Server) Stream(stream string, reply *string) error {
 }
 
 func (t *Server) CreateRule(rule *model.RPCArgDesc, reply *string) error {
-	id, err := createRule(rule.Name, rule.Json)
+	id, err := registry.CreateRule(rule.Name, rule.Json)
 	if err != nil {
 		return fmt.Errorf("Create rule %s error : %s.", id, err)
 	} else {
@@ -156,7 +158,7 @@ func (t *Server) CreateRule(rule *model.RPCArgDesc, reply *string) error {
 }
 
 func (t *Server) GetStatusRule(name string, reply *string) error {
-	if r, err := getRuleStatus(name); err != nil {
+	if r, err := registry.GetRuleStatus(name); err != nil {
 		return err
 	} else {
 		*reply = r
@@ -165,7 +167,7 @@ func (t *Server) GetStatusRule(name string, reply *string) error {
 }
 
 func (t *Server) GetTopoRule(name string, reply *string) error {
-	if r, err := getRuleTopo(name); err != nil {
+	if r, err := registry.GetRuleTopo(name); err != nil {
 		return err
 	} else {
 		dst := &bytes.Buffer{}
@@ -179,7 +181,7 @@ func (t *Server) GetTopoRule(name string, reply *string) error {
 }
 
 func (t *Server) StartRule(name string, reply *string) error {
-	if err := startRule(name); err != nil {
+	if err := registry.StartRule(name); err != nil {
 		return err
 	} else {
 		*reply = fmt.Sprintf("Rule %s was started", name)
@@ -188,12 +190,16 @@ func (t *Server) StartRule(name string, reply *string) error {
 }
 
 func (t *Server) StopRule(name string, reply *string) error {
-	*reply, _ = stopRule(name)
+	if err := registry.StopRule(name); err != nil {
+		return err
+	} else {
+		*reply = fmt.Sprintf("Rule %s was stopped.", name)
+	}
 	return nil
 }
 
 func (t *Server) RestartRule(name string, reply *string) error {
-	err := restartRule(name)
+	err := registry.RestartRule(name)
 	if err != nil {
 		return err
 	}
@@ -212,7 +218,7 @@ func (t *Server) DescRule(name string, reply *string) error {
 }
 
 func (t *Server) ShowRules(_ int, reply *string) error {
-	r, err := getAllRulesWithStatus()
+	r, err := registry.GetAllRulesWithStatus()
 	if err != nil {
 		return fmt.Errorf("Show rule error : %s.", err)
 	}
@@ -233,22 +239,16 @@ func (t *Server) ShowRules(_ int, reply *string) error {
 }
 
 func (t *Server) DropRule(name string, reply *string) error {
-	deleteRule(name)
-	r, err := ruleProcessor.ExecDrop(name)
+	err := registry.DeleteRule(name)
 	if err != nil {
 		return fmt.Errorf("Drop rule error : %s.", err)
-	} else {
-		err := t.StopRule(name, reply)
-		if err != nil {
-			return err
-		}
 	}
-	*reply = r
+	*reply = fmt.Sprintf("Rule %s is dropped.", name)
 	return nil
 }
 
 func (t *Server) ValidateRule(rule *model.RPCArgDesc, reply *string) error {
-	_, s, err := validateRule(rule.Name, rule.Json)
+	_, s, err := registry.ValidateRule(rule.Name, rule.Json)
 	if s {
 		*reply = "The rule has been successfully validated and is confirmed to be correct."
 	} else {
@@ -280,7 +280,7 @@ func (t *Server) Import(file string, reply *string) error {
 				logger.Error(ee)
 				continue
 			}
-			reply := recoverRule(rul)
+			reply := registry.RecoverRule(rul)
 			if reply != "" {
 				logger.Error(reply)
 			}
@@ -400,7 +400,7 @@ func initQuery() {
 		for {
 			<-ticker.C
 			if registry != nil {
-				if _, ok := registry.Load(QueryRuleId); !ok {
+				if _, ok := registry.load(QueryRuleId); !ok {
 					continue
 				}
 
diff --git a/internal/server/rule_manager.go b/internal/server/rule_manager.go
index 96038111..b1aecef4 100644
--- a/internal/server/rule_manager.go
+++ b/internal/server/rule_manager.go
@@ -15,7 +15,6 @@
 package server
 
 import (
-	"bytes"
 	"encoding/json"
 	"fmt"
 	"sort"
@@ -30,9 +29,7 @@ import (
 	"github.com/lf-edge/ekuiper/v2/internal/topo/planner"
 	"github.com/lf-edge/ekuiper/v2/internal/topo/rule"
 	"github.com/lf-edge/ekuiper/v2/internal/xsql"
-	"github.com/lf-edge/ekuiper/v2/pkg/cast"
 	"github.com/lf-edge/ekuiper/v2/pkg/errorx"
-	"github.com/lf-edge/ekuiper/v2/pkg/hidden"
 	"github.com/lf-edge/ekuiper/v2/pkg/infra"
 )
 
@@ -44,73 +41,86 @@ var registry *RuleRegistry
 
 type RuleRegistry struct {
 	sync.RWMutex
-	internal map[string]*rule.RuleState
+	internal map[string]*rule.State
 }
 
-// Store create the in memory entry for a rule. Run in:
-// 1. Restore the rules from KV at startup
-// 2. Restore the rules when importing
-// 3. Create a rule
-func (rr *RuleRegistry) Store(key string, value *rule.RuleState) {
-	rr.Lock()
-	rr.internal[key] = value
-	rr.Unlock()
-}
+//// registry and db level state change functions
 
-// Load the entry of a rule by id. It is used to get the current rule state
+// load the entry of a rule by id. It is used to get the current rule state
 // or send command to a running rule
-func (rr *RuleRegistry) Load(key string) (value *rule.RuleState, ok bool) {
+func (rr *RuleRegistry) load(key string) (value *rule.State, ok bool) {
 	rr.RLock()
 	result, ok := rr.internal[key]
 	rr.RUnlock()
 	return result, ok
 }
 
-// Delete Atomic get and delete. Only run when deleting a rule in runtime.
-func (rr *RuleRegistry) Delete(key string) (*rule.RuleState, bool) {
+// register and save to db
+func (rr *RuleRegistry) save(key string, ruleJson string, value *rule.State) error {
+	rr.Lock()
+	defer rr.Unlock()
+	rr.internal[key] = value
+	return ruleProcessor.ExecCreate(key, ruleJson)
+}
+
+// only register. It is called when recover from db
+func (rr *RuleRegistry) register(key string, value *rule.State) {
+	rr.Lock()
+	defer rr.Unlock()
+	rr.internal[key] = value
+}
+
+func (rr *RuleRegistry) update(id string, ruleJson string) error {
+	rr.Lock()
+	defer rr.Unlock()
+	_, err := ruleProcessor.ExecUpdate(id, ruleJson)
+	return err
+}
+
+func (rr *RuleRegistry) updateTrigger(id string, trigger bool) error {
 	rr.Lock()
+	defer rr.Unlock()
+	_, err := ruleProcessor.ExecReplaceRuleState(id, trigger)
+	return err
+}
+
+func (rr *RuleRegistry) delete(key string) (*rule.State, error) {
+	rr.Lock()
+	defer rr.Unlock()
+	var err error
 	result, ok := rr.internal[key]
 	if ok {
 		delete(rr.internal, key)
+		err = ruleProcessor.ExecDrop(key)
+	} else {
+		err = fmt.Errorf("rule %s not found", key)
 	}
-	rr.Unlock()
-	return result, ok
+	return result, err
 }
 
-func createRule(name, ruleJson string) (id string, err error) {
-	var rs *rule.RuleState = nil
+//// APIs for REST service
 
+func (rr *RuleRegistry) CreateRule(name, ruleJson string) (id string, err error) {
 	// Validate the rule json
 	r, err := ruleProcessor.GetRuleByJson(name, ruleJson)
 	if err != nil {
 		return "", fmt.Errorf("invalid rule json: %v", err)
 	}
-
-	if exists := ruleProcessor.ExecExists(r.Id); exists {
-		return r.Id, fmt.Errorf("rule %v already exists", r.Id)
+	if _, ok := rr.load(r.Id); ok {
+		return name, fmt.Errorf("rule %s already exists", r.Id)
 	}
-
+	// create state and save
+	rs := rule.NewState(r)
 	// Validate the topo
-	err = infra.SafeRun(func() error {
-		rs, err = createRuleState(r)
-		return err
-	})
+	err = rs.Validate()
 	if err != nil {
 		return r.Id, err
 	}
-	defer func() {
-		if err != nil {
-			// Do not store to registry so also delete the KV
-			deleteRule(id)
-		}
-	}()
-
-	// Store to KV
-	err = ruleProcessor.ExecCreate(r.Id, ruleJson)
+	// Store to registry and KV
+	err = rr.save(r.Id, ruleJson, rs)
 	if err != nil {
 		return r.Id, fmt.Errorf("store the rule error: %v", err)
 	}
-
 	// Start the rule asyncly
 	if r.Triggered {
 		go func() {
@@ -126,32 +136,12 @@ func createRule(name, ruleJson string) (id string, err error) {
 	return r.Id, nil
 }
 
-// Create and initialize a rule state.
-// Errors are possible during plan the topo.
-// If error happens return immediately without add it to the registry
-func createRuleState(r *def.Rule) (*rule.RuleState, error) {
-	rs, err := rule.NewRuleState(r)
-	if err != nil {
-		return rs, err
-	}
-	registry.Store(r.Id, rs)
-	return rs, nil
-}
-
-func recoverRule(r *def.Rule) string {
-	var rs *rule.RuleState = nil
-	var err error = nil
-	// Validate the topo
-	panicOrError := infra.SafeRun(func() error {
-		rs, err = createRuleState(r)
-		return err
-	})
-
-	if panicOrError != nil { // when recovering rules, assume the rules are valid, so always add it to the registry
-		conf.Log.Errorf("Create rule topo error: %v", err)
-		r.Triggered = false
-		registry.Store(r.Id, rs)
-	}
+// RecoverRule loads in imported rule.
+// Unlike creation, 1. it suppose the rule is valid thus, it will always create the rule state in registry
+// 2. It does not handle rule saving to db.
+func (rr *RuleRegistry) RecoverRule(r *def.Rule) string {
+	rs := rule.NewState(r)
+	rr.register(r.Id, rs)
 	if !r.Triggered {
 		return fmt.Sprintf("Rule %s was stopped.", r.Id)
 	} else {
@@ -166,211 +156,241 @@ func recoverRule(r *def.Rule) string {
 	return fmt.Sprintf("Rule %s was started.", r.Id)
 }
 
-// reload password from resources if the config both include password(as fake password) and resourceId
-func replacePasswdForConfig(typ string, name string, config map[string]interface{}) map[string]interface{} {
-	if r, ok := config["resourceId"]; ok {
-		if resourceId, ok := r.(string); ok {
-			return meta.ReplacePasswdForConfig(typ, name, resourceId, config)
-		}
-	}
-	return config
-}
-
-func replaceRulePassword(id, ruleJson string) (string, error) {
-	r := &def.Rule{
-		Triggered: true,
+// UpdateRule validates the new rule, then update the db, then restart the rule
+func (rr *RuleRegistry) UpdateRule(ruleId, ruleJson string) error {
+	// Validate the rule json
+	r, err := ruleProcessor.GetRuleByJson(ruleId, ruleJson)
+	if err != nil {
+		return fmt.Errorf("Invalid rule json: %v", err)
 	}
-	if err := json.Unmarshal([]byte(ruleJson), r); err != nil {
-		return "", err
+	rs, ok := registry.load(ruleId)
+	if !ok {
+		return errorx.NewWithCode(errorx.NOT_FOUND, fmt.Sprintf("Rule %s is not found in registry, please check if it is created", ruleId))
 	}
-	existsRule, err := ruleProcessor.GetRuleById(id)
+	// Try plan with the new json. If err, revert to old rule
+	oldRule := rs.Rule
+	rs.Rule = r
+	err = rs.Validate()
 	if err != nil {
-		return "", err
+		rs.Rule = oldRule
+		return err
 	}
-
-	var replacePassword bool
-	for i, action := range r.Actions {
-		if i >= len(existsRule.Actions) {
-			break
-		}
-		for k, v := range action {
-			if m, ok := v.(map[string]interface{}); ok {
-				for key := range hidden.GetHiddenKeys() {
-					if v, ok := m[key]; ok && v == hidden.PASSWORD {
-						oldAction := existsRule.Actions[i]
-						oldV, ok := oldAction[k]
-						if ok {
-							if oldM, ok := oldV.(map[string]interface{}); ok {
-								oldPasswordValue, ok := oldM[key]
-								if ok {
-									oldPasswordStr, ok := oldPasswordValue.(string)
-									if ok && oldPasswordStr != hidden.PASSWORD {
-										m[key] = oldPasswordStr
-										action[k] = m
-										r.Actions[i] = action
-										replacePassword = true
-										continue
-									}
-								}
-							}
-						}
-					}
-				}
-			}
+	// Validate successful, save to db
+	err1 := rr.update(r.Id, ruleJson)
+	// ReRun the rule
+	rs.Stop()
+	if r.Triggered {
+		err2 := rs.Start()
+		if err2 != nil {
+			return err2
 		}
 	}
-	if !replacePassword {
-		return ruleJson, nil
-	}
-	b, err := json.Marshal(r)
-	if err != nil {
-		return "", err
-	}
-	return string(b), nil
+	return err1
 }
 
-func updateRule(ruleId, ruleJson string, replacePasswd bool) error {
-	// Validate the rule json
-	r, err := ruleProcessor.GetRuleByJson(ruleId, ruleJson)
-	if err != nil {
-		return fmt.Errorf("Invalid rule json: %v", err)
-	}
-	if replacePasswd {
-		for i, action := range r.Actions {
-			for k, v := range action {
-				if m, ok := v.(map[string]interface{}); ok {
-					m = replacePasswdForConfig("sink", k, m)
-					action[k] = m
-				}
-			}
-			r.Actions[i] = action
+func (rr *RuleRegistry) DeleteRule(name string) error {
+	// lock registry and db. rs level has its own lock
+	rs, err := rr.delete(name)
+	if rs != nil {
+		err = rs.Delete()
+		if err != nil {
+			logger.Errorf("delete rule %s error: %v", name, err)
 		}
+		deleteRuleMetrics(name)
 	}
-	if rs, ok := registry.Load(r.Id); ok {
-		err := rs.UpdateTopo(r)
+	return err
+}
+
+func (rr *RuleRegistry) StartRule(name string) error {
+	rs, ok := registry.load(name)
+	if !ok {
+		return errorx.NewWithCode(errorx.NOT_FOUND, fmt.Sprintf("Rule %s is not found in registry, please check if it is created", name))
+	} else {
+		err := rr.updateTrigger(name, true)
 		if err != nil {
-			return err
+			conf.Log.Warnf("start rule update db status error: %s", err.Error())
 		}
-		_, err = ruleProcessor.ExecReplaceRuleState(rs.RuleId, r.Triggered)
-		return err
-	} else {
-		return fmt.Errorf("Rule %s registry not found, try to delete it and recreate", r.Id)
+		return rs.Start()
 	}
 }
 
-func deleteRule(name string) (result string) {
-	if rs, ok := registry.Delete(name); ok {
-		rs.Close()
-		deleteRuleMetrics(name)
-		result = fmt.Sprintf("Rule %s was deleted.", name)
+func (rr *RuleRegistry) StopRule(name string) error {
+	if rs, ok := registry.load(name); ok {
+		err := rr.updateTrigger(name, false)
+		if err != nil {
+			conf.Log.Warnf("stop rule update db status error: %s", err.Error())
+		}
+		rs.Stop()
 	} else {
-		result = fmt.Sprintf("Rule %s was not found.", name)
+		return errorx.NewWithCode(errorx.NOT_FOUND, fmt.Sprintf("Rule %s is not found in registry, please check if it is created", name))
 	}
-	return
+	return nil
 }
 
-func deleteRuleMetrics(name string) {
-	if conf.Config != nil && conf.Config.Basic.Prometheus {
-		promMetrics.RemoveRuleStatus(name)
+func (rr *RuleRegistry) RestartRule(name string) error {
+	if rs, ok := registry.load(name); ok {
+		err := rr.updateTrigger(name, true)
+		if err != nil {
+			conf.Log.Warnf("restart rule update db status error: %s", err.Error())
+		}
+		rs.Stop()
+		return rs.Start()
+	} else {
+		return errorx.NewWithCode(errorx.NOT_FOUND, fmt.Sprintf("Rule %s is not found in registry, please check if it is created", name))
 	}
 }
 
-func startRule(name string) error {
-	return reRunRule(name, false)
+func (rr *RuleRegistry) GetAllRuleStatus() (string, error) {
+	rules, err := ruleProcessor.GetAllRules()
+	if err != nil {
+		return "", err
+	}
+	m := make(map[string]ruleExceptionStatus)
+	for _, ruleID := range rules {
+		s, err := getRuleExceptionStatus(ruleID)
+		if err != nil {
+			return "", err
+		}
+		m[ruleID] = s
+	}
+	b, _ := json.Marshal(m)
+	return string(b), nil
 }
 
-func startRuleInternal(name string) error {
-	return reRunRule(name, true)
+func (rr *RuleRegistry) GetAllRulesWithStatus() ([]map[string]any, error) {
+	ruleIds, err := ruleProcessor.GetAllRules()
+	if err != nil {
+		return nil, err
+	}
+	sort.Strings(ruleIds)
+	result := make([]map[string]interface{}, len(ruleIds))
+	for i, id := range ruleIds {
+		ruleName := id
+		ruleDef, _ := ruleProcessor.GetRuleById(id)
+		if ruleDef != nil && ruleDef.Name != "" {
+			ruleName = ruleDef.Name
+		}
+		var str string
+		s, err := getRuleState(id)
+		if err != nil {
+			str = fmt.Sprintf("error: %s", err)
+		} else {
+			str = rule.StateName[s]
+		}
+		result[i] = map[string]interface{}{
+			"id":     id,
+			"name":   ruleName,
+			"status": str,
+		}
+	}
+	return result, nil
 }
 
-// reRunRule rerun the rule from optimize to Open the operator in order to refresh the schema
-func reRunRule(name string, isInternal bool) error {
-	rs, ok := registry.Load(name)
-	if !ok {
-		return errorx.NewWithCode(errorx.NOT_FOUND, fmt.Sprintf("Rule %s is not found in registry, please check if it is created", name))
+func (rr *RuleRegistry) GetRuleStatus(name string) (string, error) {
+	if rs, ok := registry.load(name); ok {
+		return rs.GetStatusMessage(), nil
 	} else {
-		if !isInternal {
-			if rule, err := ruleProcessor.ExecReplaceRuleState(rs.RuleId, true); err != nil {
-				return err
-			} else {
-				rs.Rule = rule
-			}
-		}
-		return rs.UpdateTopo(rs.Rule)
+		return "", errorx.NewWithCode(errorx.NOT_FOUND, fmt.Sprintf("Rule %s is not found", name))
 	}
 }
 
-func stopRuleInternal(name string) {
-	var err error
-	if rs, ok := registry.Load(name); ok {
-		err = rs.InternalStop()
-		if err != nil {
-			conf.Log.Warn(err)
-		}
+func (rr *RuleRegistry) GetRuleStatusV2(name string) (map[string]any, error) {
+	if rs, ok := rr.load(name); ok {
+		return rs.GetStatusMap(), nil
+	} else {
+		return nil, errorx.NewWithCode(errorx.NOT_FOUND, fmt.Sprintf("Rule %s is not found", name))
 	}
 }
 
-func stopRuleWhenServerStop(name string) {
-	var err error
-	if rs, ok := registry.Load(name); ok {
-		err = rs.Stop()
+func (rr *RuleRegistry) GetRuleTopo(name string) (string, error) {
+	if rs, ok := registry.load(name); ok {
+		graph := rs.GetTopoGraph()
+		if graph == nil {
+			return "", errorx.New(fmt.Sprintf("Fail to get rule %s's topo, make sure the rule has been started before", name))
+		}
+		bs, err := json.Marshal(graph)
 		if err != nil {
-			conf.Log.Warn(err)
+			return "", errorx.New(fmt.Sprintf("Fail to encode rule %s's topo", name))
+		} else {
+			return string(bs), nil
 		}
+	} else {
+		return "", errorx.NewWithCode(errorx.NOT_FOUND, fmt.Sprintf("Rule %s is not found", name))
 	}
 }
 
-func stopRule(name string) (result string, err error) {
-	if rs, ok := registry.Load(name); ok {
-		err = rs.Stop()
+func (rr *RuleRegistry) ValidateRule(name, ruleJson string) ([]string, bool, error) {
+	// Validate the ruleDef json
+	ruleDef, err := ruleProcessor.GetRuleByJson(name, ruleJson)
+	if err != nil {
+		return nil, false, fmt.Errorf("invalid rule json: %v", err)
+	}
+	var sources []string
+	if len(ruleDef.Sql) > 0 {
+		stmt, _ := xsql.GetStatementFromSql(ruleDef.Sql)
+		s, err := store.GetKV("stream")
 		if err != nil {
-			conf.Log.Warn(err)
+			return nil, false, err
 		}
-		_, err = ruleProcessor.ExecReplaceRuleState(name, false)
+		sources = xsql.GetStreams(stmt)
+		for _, result := range sources {
+			_, err := xsql.GetDataSource(s, result)
+			if err != nil {
+				return nil, false, err
+			}
+		}
+	} else if ruleDef.Graph != nil {
+		tp, err := planner.PlanByGraph(ruleDef)
 		if err != nil {
-			conf.Log.Warnf("stop rule found error: %s", err.Error())
+			return nil, false, fmt.Errorf("invalid ruleDef graph: %v", err)
 		}
-		result = fmt.Sprintf("Rule %s was stopped.", name)
-	} else {
-		result = fmt.Sprintf("Rule %s was not found.", name)
-		err = errorx.NewWithCode(errorx.NOT_FOUND, result)
+		sources = tp.GetTopo().Sources
 	}
-	return
+	return sources, true, nil
 }
 
-func restartRule(name string) error {
-	return reRunRule(name, false)
+/// Rule Scheduler internal API
+
+func (rr *RuleRegistry) scheduledStart(name string) error {
+	rs, ok := registry.load(name)
+	if !ok {
+		return errorx.NewWithCode(errorx.NOT_FOUND, fmt.Sprintf("Scheduled rule %s is not found in registry, please check if it is deleted", name))
+	} else {
+		return rs.ScheduleStart()
+	}
 }
 
-func getAllRuleStatus() (string, error) {
-	rules, err := ruleProcessor.GetAllRules()
-	if err != nil {
-		return "", err
+func (rr *RuleRegistry) scheduledStop(name string) error {
+	rs, ok := registry.load(name)
+	if !ok {
+		return errorx.NewWithCode(errorx.NOT_FOUND, fmt.Sprintf("Scheduled rule %s is not found in registry, please check if it is deleted", name))
+	} else {
+		rs.ScheduleStop()
+		return nil
 	}
-	m := make(map[string]ruleExceptionStatus)
-	for _, ruleID := range rules {
-		s, err := getRuleExceptionStatus(ruleID)
-		if err != nil {
-			return "", err
-		}
-		m[ruleID] = s
+}
+
+func (rr *RuleRegistry) stopAtExit(name string) error {
+	rs, ok := registry.load(name)
+	if !ok {
+		return errorx.NewWithCode(errorx.NOT_FOUND, fmt.Sprintf("Rule %s is not found in registry, please check if it is deleted", name))
+	} else {
+		rs.Stop()
+		return nil
 	}
-	b, _ := json.Marshal(m)
-	return string(b), nil
 }
 
+//// Util functions
+
 func getRuleExceptionStatus(name string) (ruleExceptionStatus, error) {
 	s := ruleExceptionStatus{
 		lastExceptionTime: -1,
 	}
-	if rs, ok := registry.Load(name); ok {
-		result, err := rs.GetState()
-		if err != nil {
-			return s, err
-		}
-		s.Status = result
-		if result == rule.RuleStarted {
-			keys, values := (*rs.Topology).GetMetrics()
+	if rs, ok := registry.load(name); ok {
+		st := rs.GetState()
+		s.Status = rule.StateName[st]
+		if st == rule.Running {
+			keys, values := rs.GetMetrics()
 			for i, key := range keys {
 				if strings.Contains(key, "last_exception_time") {
 					v := values[i].(int64)
@@ -410,110 +430,9 @@ type ruleExceptionStatus struct {
 	lastExceptionTime int64
 }
 
-func getRuleStatusV2(name string) (map[string]any, error) {
-	if rs, ok := registry.Load(name); ok {
-		result, _ := rs.GetState()
-		metrics := make(map[string]any)
-		if result == "Running" {
-			opMetrics := (*rs.Topology).GetMetricsV2()
-			metrics["status"] = "running"
-			lastStart, lastStop, nextStart := rs.GetScheduleTimestamp()
-			metrics["lastStartTimestamp"] = lastStart
-			metrics["lastStopTimestamp"] = lastStop
-			metrics["nextStopTimestamp"] = nextStart
-			for key, m := range opMetrics {
-				metrics[key] = m
-			}
-		} else {
-			metrics["status"] = "stopped"
-			metrics["message"] = result
-		}
-		return metrics, nil
-	} else {
-		return nil, errorx.NewWithCode(errorx.NOT_FOUND, fmt.Sprintf("Rule %s is not found", name))
-	}
-}
-
-func getRuleStatus(name string) (string, error) {
-	if rs, ok := registry.Load(name); ok {
-		result, err := rs.GetState()
-		if err != nil {
-			return "", err
-		}
-		if result == "Running" {
-			keys, values := (*rs.Topology).GetMetrics()
-			metrics := "{"
-			metrics += `"status": "running",`
-			lastStart, lastStop, nextStart := rs.GetScheduleTimestamp()
-			metrics += fmt.Sprintf(`"lastStartTimestamp": "%v",`, lastStart)
-			metrics += fmt.Sprintf(`"lastStopTimestamp": "%v",`, lastStop)
-			metrics += fmt.Sprintf(`"nextStopTimestamp": "%v",`, nextStart)
-			for i, key := range keys {
-				value := values[i]
-				switch value.(type) {
-				case string:
-					metrics += fmt.Sprintf("\"%s\":%q,", key, value)
-				default:
-					metrics += fmt.Sprintf("\"%s\":%v,", key, value)
-				}
-			}
-			metrics = metrics[:len(metrics)-1] + "}"
-			dst := &bytes.Buffer{}
-			if err = json.Indent(dst, cast.StringToBytes(metrics), "", "  "); err != nil {
-				result = metrics
-			} else {
-				result = dst.String()
-			}
-		} else {
-			return getStoppedState(result)
-		}
-		return result, nil
-	} else {
-		return "", errorx.NewWithCode(errorx.NOT_FOUND, fmt.Sprintf("Rule %s is not found", name))
-	}
-}
-
-func getStoppedState(message string) (string, error) {
-	s := map[string]string{
-		"status":  "stopped",
-		"message": message,
-	}
-	re, err := json.Marshal(s)
-	if err != nil {
-		return "", err
-	}
-	return string(re), nil
-}
-
-func getAllRulesWithStatus() ([]map[string]interface{}, error) {
-	ruleIds, err := ruleProcessor.GetAllRules()
-	if err != nil {
-		return nil, err
-	}
-	sort.Strings(ruleIds)
-	result := make([]map[string]interface{}, len(ruleIds))
-	for i, id := range ruleIds {
-		ruleName := id
-		rule, _ := ruleProcessor.GetRuleById(id)
-		if rule != nil && rule.Name != "" {
-			ruleName = rule.Name
-		}
-		s, err := getRuleState(id)
-		if err != nil {
-			s = fmt.Sprintf("error: %s", err)
-		}
-		result[i] = map[string]interface{}{
-			"id":     id,
-			"name":   ruleName,
-			"status": s,
-		}
-	}
-	return result, nil
-}
-
 type ruleWrapper struct {
 	rule  *def.Rule
-	state string
+	state rule.RunState
 }
 
 func getAllRulesWithState() ([]ruleWrapper, error) {
@@ -524,66 +443,35 @@ func getAllRulesWithState() ([]ruleWrapper, error) {
 	sort.Strings(ruleIds)
 	rules := make([]ruleWrapper, 0, len(ruleIds))
 	for _, id := range ruleIds {
-		rs, ok := registry.Load(id)
+		rs, ok := registry.load(id)
 		if ok {
-			s, _ := rs.GetState()
+			s := rs.GetState()
 			rules = append(rules, ruleWrapper{rule: rs.Rule, state: s})
 		}
 	}
 	return rules, nil
 }
 
-func getRuleState(name string) (string, error) {
-	if rs, ok := registry.Load(name); ok {
-		return rs.GetState()
+func getRuleState(name string) (rule.RunState, error) {
+	if rs, ok := registry.load(name); ok {
+		return rs.GetState(), nil
 	} else {
-		return "", fmt.Errorf("Rule %s is not found in registry", name)
+		return rule.Stopped, fmt.Errorf("Rule %s is not found in registry", name)
 	}
 }
 
-func getRuleTopo(name string) (string, error) {
-	if rs, ok := registry.Load(name); ok {
-		graph := rs.GetTopoGraph()
-		if graph == nil {
-			return "", errorx.New(fmt.Sprintf("Fail to get rule %s's topo, make sure the rule has been started before", name))
-		}
-		bs, err := json.Marshal(graph)
-		if err != nil {
-			return "", errorx.New(fmt.Sprintf("Fail to encode rule %s's topo", name))
-		} else {
-			return string(bs), nil
-		}
-	} else {
-		return "", errorx.NewWithCode(errorx.NOT_FOUND, fmt.Sprintf("Rule %s is not found", name))
+func deleteRuleMetrics(name string) {
+	if conf.Config != nil && conf.Config.Basic.Prometheus {
+		promMetrics.RemoveRuleStatus(name)
 	}
 }
 
-func validateRule(name, ruleJson string) ([]string, bool, error) {
-	// Validate the rule json
-	rule, err := ruleProcessor.GetRuleByJson(name, ruleJson)
-	if err != nil {
-		return nil, false, fmt.Errorf("invalid rule json: %v", err)
-	}
-	var sources []string
-	if len(rule.Sql) > 0 {
-		stmt, _ := xsql.GetStatementFromSql(rule.Sql)
-		s, err := store.GetKV("stream")
-		if err != nil {
-			return nil, false, err
-		}
-		sources = xsql.GetStreams(stmt)
-		for _, result := range sources {
-			_, err := xsql.GetDataSource(s, result)
-			if err != nil {
-				return nil, false, err
-			}
-		}
-	} else if rule.Graph != nil {
-		tp, err := planner.PlanByGraph(rule)
-		if err != nil {
-			return nil, false, fmt.Errorf("invalid rule graph: %v", err)
+// reload password from resources if the config both include password(as fake password) and resourceId
+func replacePasswdForConfig(typ string, name string, config map[string]interface{}) map[string]interface{} {
+	if r, ok := config["resourceId"]; ok {
+		if resourceId, ok := r.(string); ok {
+			return meta.ReplacePasswdForConfig(typ, name, resourceId, config)
 		}
-		sources = tp.GetTopo().Sources
 	}
-	return sources, true, nil
+	return config
 }
diff --git a/internal/server/rule_state.go b/internal/server/rule_state.go
index 75bb6060..5c141dea 100644
--- a/internal/server/rule_state.go
+++ b/internal/server/rule_state.go
@@ -79,16 +79,16 @@ func updateRuleOffset(ruleID string, param map[string]interface{}) error {
 		switch val.(int) {
 		case 1:
 			StateErr = nil
-			s = rule.RuleStarted
+			s = rule.Running
 		case 2:
 			StateErr = nil
-			s = rule.RuleStopped
+			s = rule.Stopped
 		}
 	})
 	if StateErr != nil {
 		return StateErr
 	}
-	if s != rule.RuleStarted {
+	if s != rule.Running {
 		return fmt.Errorf("rule %v should be running when modify state", ruleID)
 	}
 
@@ -96,9 +96,9 @@ func updateRuleOffset(ruleID string, param map[string]interface{}) error {
 	if err := cast.MapToStruct(param, req); err != nil {
 		return err
 	}
-	rs, ok := registry.Load(ruleID)
+	rs, ok := registry.load(ruleID)
 	if !ok {
 		return fmt.Errorf("rule %s is not found in registry", ruleID)
 	}
-	return rs.Topology.ResetStreamOffset(req.StreamName, req.Input)
+	return rs.ResetStreamOffset(req.StreamName, req.Input)
 }
diff --git a/internal/server/server.go b/internal/server/server.go
index 6c4a8a5d..f9900290 100644
--- a/internal/server/server.go
+++ b/internal/server/server.go
@@ -201,7 +201,7 @@ func StartUp(Version string) {
 	meta.Bind()
 	initRuleset()
 
-	registry = &RuleRegistry{internal: make(map[string]*rule.RuleState)}
+	registry = &RuleRegistry{internal: make(map[string]*rule.State)}
 	// Start lookup tables
 	streamProcessor.RecoverLookupTable()
 	// Start rules
@@ -217,7 +217,7 @@ func StartUp(Version string) {
 				continue
 			}
 			// err = server.StartRule(rule, &reply)
-			reply = recoverRule(rule)
+			reply = registry.RecoverRule(rule)
 			if 0 != len(reply) {
 				logger.Info(reply)
 			}
@@ -343,8 +343,7 @@ func resetAllRules() error {
 		return err
 	}
 	for _, name := range rules {
-		_ = deleteRule(name)
-		_, err := ruleProcessor.ExecDrop(name)
+		err := registry.DeleteRule(name)
 		if err != nil {
 			logger.Warnf("delete rule: %s with error %v", name, err)
 			continue
@@ -422,15 +421,15 @@ func handleAllRuleStatusMetrics(rs []ruleWrapper) {
 		for _, r := range rs {
 			id := r.rule.Id
 			switch r.state {
-			case rule.RuleStarted:
+			case rule.Running:
 				runningCount++
 				v = RuleRunning
-			case rule.RuleStopped, rule.RuleTerminated, rule.RuleWait:
+			case rule.StoppedByErr:
 				stopCount++
-				v = RuleStopped
+				v = RuleStoppedByError
 			default:
 				stopCount++
-				v = RuleStoppedByError
+				v = RuleStopped
 			}
 			promMetrics.SetRuleStatus(id, int(v))
 		}
@@ -441,20 +440,22 @@ func handleAllRuleStatusMetrics(rs []ruleWrapper) {
 
 func handleAllScheduleRuleState(now time.Time, rs []ruleWrapper) {
 	for _, r := range rs {
-		if err := handleScheduleRuleState(now, r.rule, r.state); err != nil {
+		if err := handleScheduleRuleState(now, r.rule); err != nil {
 			conf.Log.Errorf("handle schedule rule %v state failed, err:%v", r.rule.Id, err)
 		}
 	}
 }
 
-func handleScheduleRuleState(now time.Time, r *def.Rule, state string) error {
-	scheduleActionSignal := handleScheduleRule(now, r, state)
-	conf.Log.Debugf("rule %v origin state: %v, sginal: %v", r.Id, state, scheduleActionSignal)
+func handleScheduleRuleState(now time.Time, r *def.Rule) error {
+	scheduleActionSignal := handleScheduleRule(now, r)
+	conf.Log.Debugf("rule %v, sginal: %v", r.Id, scheduleActionSignal)
 	switch scheduleActionSignal {
 	case scheduleRuleActionStart:
-		return startRuleInternal(r.Id)
+		return registry.scheduledStart(r.Id)
 	case scheduleRuleActionStop:
-		stopRuleInternal(r.Id)
+		return registry.scheduledStop(r.Id)
+	default:
+		// do nothing
 	}
 	return nil
 }
@@ -467,7 +468,7 @@ const (
 	scheduleRuleActionStop
 )
 
-func handleScheduleRule(now time.Time, r *def.Rule, state string) scheduleRuleAction {
+func handleScheduleRule(now time.Time, r *def.Rule) scheduleRuleAction {
 	options := r.Options
 	if options != nil && options.Cron == "" && options.Duration == "" && len(options.CronDatetimeRange) > 0 {
 		isInRange, err := schedule.IsInScheduleRanges(now, options.CronDatetimeRange)
@@ -475,9 +476,9 @@ func handleScheduleRule(now time.Time, r *def.Rule, state string) scheduleRuleAc
 			conf.Log.Errorf("check rule %v schedule failed, err:%v", r.Id, err)
 			return scheduleRuleActionDoNothing
 		}
-		if isInRange && state == rule.RuleWait && r.Triggered {
+		if isInRange && r.Triggered {
 			return scheduleRuleActionStart
-		} else if !isInRange && state == rule.RuleStarted && r.Triggered {
+		} else if !isInRange && r.Triggered {
 			return scheduleRuleActionStop
 		}
 	}
@@ -507,6 +508,9 @@ func startCPUProfiling(ctx context.Context) error {
 func waitAllRuleStop() {
 	rules, _ := ruleProcessor.GetAllRules()
 	for _, r := range rules {
-		stopRuleWhenServerStop(r)
+		err := registry.stopAtExit(r)
+		if err != nil {
+			logger.Warnf("stop rule %s failed, err:%v", r, err)
+		}
 	}
 }
diff --git a/internal/server/yaml_import_export.go b/internal/server/yaml_import_export.go
index bd18b5fb..dd151721 100644
--- a/internal/server/yaml_import_export.go
+++ b/internal/server/yaml_import_export.go
@@ -488,10 +488,9 @@ func importNativePlugins(m *MetaConfiguration) error {
 
 func importRules(m *MetaConfiguration) error {
 	for key, value := range m.Rules {
-		deleteRule(key)
-		ruleProcessor.ExecDrop(key)
+		_ = registry.DeleteRule(key)
 		b, _ := json.Marshal(value)
-		_, err := createRule(key, string(b)) //nolint:staticcheck
+		_, err := registry.CreateRule(key, string(b)) //nolint:staticcheck
 		err = mockImportErr(err, mockRulesErr)
 		if err != nil {
 			return fmt.Errorf("replace rule %v failed, err:%v", key, err)
diff --git a/internal/topo/rule/cron.go b/internal/topo/rule/cron.go
new file mode 100644
index 00000000..c99c08c8
--- /dev/null
+++ b/internal/topo/rule/cron.go
@@ -0,0 +1,187 @@
+// Copyright 2024 EMQ Technologies Co., Ltd.
+//
+// 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 rule
+
+import (
+	"context"
+	"fmt"
+	"strings"
+	"time"
+
+	"github.com/robfig/cron/v3"
+
+	"github.com/lf-edge/ekuiper/v2/internal/conf"
+	"github.com/lf-edge/ekuiper/v2/internal/pkg/schedule"
+	"github.com/lf-edge/ekuiper/v2/pkg/timex"
+)
+
+type cronInterface interface {
+	Start()
+	AddFunc(spec string, cmd func()) (cron.EntryID, error)
+	Remove(id cron.EntryID)
+}
+
+var backgroundCron cronInterface
+
+func init() {
+	if !conf.IsTesting {
+		backgroundCron = cron.New()
+	} else {
+		backgroundCron = &MockCron{}
+	}
+	backgroundCron.Start()
+}
+
+type cronStateCtx struct {
+	cancel  context.CancelFunc
+	entryID cron.EntryID
+	// isInSchedule indicates the current Rule is in scheduled in backgroundCron
+	isInSchedule   bool
+	startFailedCnt int
+
+	// only used for test
+	cron     string
+	duration string
+}
+
+func (s *State) isInAllowedTimeRange(now time.Time) (bool, error) {
+	return schedule.IsInScheduleRanges(now, s.Rule.Options.CronDatetimeRange)
+}
+
+func (s *State) getNextScheduleStartTime() int64 {
+	if s.Rule.IsScheduleRule() && len(s.Rule.Options.Cron) > 0 {
+		isIn, err := schedule.IsInScheduleRanges(timex.GetNow(), s.Rule.Options.CronDatetimeRange)
+		if err == nil && isIn {
+			s, err := cron.ParseStandard(s.Rule.Options.Cron)
+			if err == nil {
+				return s.Next(timex.GetNow()).UnixMilli()
+			}
+		}
+	}
+	return 0
+}
+
+func (s *State) isInRunningSchedule(now time.Time, d time.Duration) (bool, time.Duration, error) {
+	allowed, err := s.isInAllowedTimeRange(now)
+	if err != nil {
+		return false, 0, err
+	}
+	if !allowed {
+		return false, 0, nil
+	}
+	cronExpr := s.Rule.Options.Cron
+	if strings.HasPrefix(cronExpr, "mock") {
+		return false, 0, nil
+	}
+	return schedule.IsInRunningSchedule(cronExpr, now, d)
+}
+
+// registerScheduleRule will register the job in the backgroundCron to run.
+// Job will do following 2 things:
+// 1. start the Rule in cron if else the job is already stopped
+// 2. after the Rule started, start an extract goroutine to stop the Rule after specific duration
+func (s *State) registerScheduleRule() error {
+	if s.cronState.isInSchedule {
+		return fmt.Errorf("Rule %s is already in schedule", s.Rule.Id)
+	}
+	d, err := time.ParseDuration(s.Rule.Options.Duration)
+	if err != nil {
+		return err
+	}
+	var cronCtx context.Context
+	cronCtx, s.cronState.cancel = context.WithCancel(context.Background())
+	now := timex.GetNow()
+	isInRunningSchedule, remainedDuration, err := s.isInRunningSchedule(now, d)
+	if err != nil {
+		return err
+	}
+	if isInRunningSchedule {
+		if err := s.runScheduleRule(); err != nil {
+			return err
+		}
+		s.stopAfterDuration(remainedDuration, cronCtx)
+	}
+	entryID, err := backgroundCron.AddFunc(s.Rule.Options.Cron, func() {
+		var started bool
+		var err error
+		if started, err = func() (bool, error) {
+			switch backgroundCron.(type) {
+			case *MockCron:
+				// skip mutex if this is a unit test
+			default:
+				s.Lock()
+				defer s.Unlock()
+			}
+			now := timex.GetNow()
+			allowed, err := s.isInAllowedTimeRange(now)
+			if err != nil {
+				return false, err
+			}
+			if !allowed {
+				return false, nil
+			}
+
+			s.cronState.cron = s.Rule.Options.Cron
+			s.cronState.duration = s.Rule.Options.Duration
+			return true, s.ScheduleStart()
+		}(); err != nil {
+			s.Lock()
+			s.cronState.startFailedCnt++
+			s.Unlock()
+			conf.Log.Errorf(err.Error())
+			return
+		}
+		if started {
+			s.stopAfterDuration(d, cronCtx)
+		}
+	})
+	if err != nil {
+		return err
+	}
+	s.cronState.isInSchedule = true
+	s.cronState.entryID = entryID
+	return nil
+}
+
+func (s *State) removeScheduleRule() {
+	if s.Rule.IsScheduleRule() && s.cronState.isInSchedule {
+		s.cronState.isInSchedule = false
+		if s.cronState.cancel != nil {
+			s.cronState.cancel()
+		}
+		s.cronState.startFailedCnt = 0
+		backgroundCron.Remove(s.cronState.entryID)
+	}
+}
+
+func (s *State) runScheduleRule() error {
+	s.cronState.cron = s.Rule.Options.Cron
+	s.cronState.duration = s.Rule.Options.Duration
+	return s.ScheduleStart()
+}
+
+// stopAfterDuration only for schedule Rule
+func (s *State) stopAfterDuration(d time.Duration, cronCtx context.Context) {
+	after := time.After(d)
+	go func(ctx context.Context) {
+		select {
+		case <-after:
+			s.ScheduleStop()
+			return
+		case <-cronCtx.Done():
+			return
+		}
+	}(cronCtx)
+}
diff --git a/internal/topo/rule/ruleState.go b/internal/topo/rule/ruleState.go
index fe8d7591..38222587 100644
--- a/internal/topo/rule/ruleState.go
+++ b/internal/topo/rule/ruleState.go
@@ -1,4 +1,4 @@
-// Copyright 2022-2024 EMQ Technologies Co., Ltd.
+// Copyright 2024 EMQ Technologies Co., Ltd.
 //
 // Licensed under the Apache License, Version 2.0 (the "License");
 // you may not use this file except in compliance with the License.
@@ -15,545 +15,545 @@
 package rule
 
 import (
+	"bytes"
 	"context"
+	"encoding/json"
+	"errors"
 	"fmt"
 	"math"
-	"math/rand"
+	"math/rand/v2"
+	"strconv"
 	"strings"
 	"sync"
 	"time"
 
-	"github.com/robfig/cron/v3"
-
+	"github.com/lf-edge/ekuiper/contract/v2/api"
 	"github.com/lf-edge/ekuiper/v2/internal/conf"
 	"github.com/lf-edge/ekuiper/v2/internal/pkg/def"
 	"github.com/lf-edge/ekuiper/v2/internal/pkg/schedule"
 	"github.com/lf-edge/ekuiper/v2/internal/topo"
 	"github.com/lf-edge/ekuiper/v2/internal/topo/planner"
+	"github.com/lf-edge/ekuiper/v2/pkg/cast"
 	"github.com/lf-edge/ekuiper/v2/pkg/errorx"
 	"github.com/lf-edge/ekuiper/v2/pkg/infra"
 	"github.com/lf-edge/ekuiper/v2/pkg/timex"
 )
 
-const (
-	RuleStarted    = "Running"
-	RuleStopped    = "Stopped: canceled manually."
-	RuleTerminated = "Stopped: schedule terminated."
-	RuleWait       = "Stopped: waiting for next schedule."
-)
-
 type ActionSignal int
 
 const (
 	ActionSignalStart ActionSignal = iota
 	ActionSignalStop
+	ActionSignalScheduledStart
+	ActionSignalScheduledStop
 )
 
-type cronInterface interface {
-	Start()
-	AddFunc(spec string, cmd func()) (cron.EntryID, error)
-	Remove(id cron.EntryID)
-}
-
-var backgroundCron cronInterface
-
-func init() {
-	if !conf.IsTesting {
-		backgroundCron = cron.New()
-	} else {
-		backgroundCron = &MockCron{}
-	}
-	backgroundCron.Start()
-}
+type RunState int
 
-type cronStateCtx struct {
-	cancel  context.CancelFunc
-	entryID cron.EntryID
-	// isInSchedule indicates the current rule is in scheduled in backgroundCron
-	isInSchedule   bool
-	startFailedCnt int
+const (
+	Stopped RunState = iota
+	Starting
+	Running
+	Stopping
+	ScheduledStop
+	StoppedByErr
+)
 
-	// only used for test
-	cron     string
-	duration string
+var StateName = map[RunState]string{
+	Stopped:       "Stopped", // normal stop and schedule terminated are here
+	Starting:      "Starting",
+	Running:       "Running",
+	Stopping:      "Stopping",
+	ScheduledStop: "Stopped: waiting for next schedule.",
+	StoppedByErr:  "Stopped by error",
 }
 
-/*********
- *  RuleState is created for each rule. Each ruleState runs two loops:
- *  1. action event loop to accept commands, such as start, stop, getStatus, delete
- *  2. topo running loop
- *  Both loops need to access the status, so lock is needed
- */
-
-type RuleState struct {
-	// Constant, never change. Channel to send signals to manage connection retry. When deleting the rule, close it.
-	RuleId   string
-	ActionCh chan ActionSignal
-	// Nearly constant, only change when update the rule
-	Rule *def.Rule
-	// States, create through rule in each rule start
-	Topology *topo.Topo
-	// 0 stop, 1 running, -1 delete, 2 internal stop, changed in actions
-	triggered int
-	// temporary storage for topo graph to make sure even rule close, the graph is still available
-	topoGraph *def.PrintableTopo
+// State control the Rule RunState
+// Created when loading from DB or creating. Deleted when Rule deleting
+// May be accessed by multiple go routines, receiving concurrent request to change the RunState
+type State struct {
 	sync.RWMutex
+	// Nearly constant, only change when update the Rule
+	// It is used to construct topo
+	Rule   *def.Rule
+	logger api.Logger
+	// concurrent running states
+	currentState RunState
+	actionQ      []ActionSignal
+	// Sync RunState, do not call in concurrent go routines
+	topology    *topo.Topo
+	cancelRetry context.CancelFunc
+	// temporary storage for topo graph to make sure even Rule close, the graph is still available
+	topoGraph *def.PrintableTopo
 	cronState cronStateCtx
-
+	// Metric RunState
 	lastStartTimestamp int64
 	lastStopTimestamp  int64
-	isClosed           bool
+	lastWill           string
+	stoppedMetrics     []any
 }
 
-// NewRuleState Create and initialize a rule state.
-// Errors are possible during plan the topo.
-// If error happens return immediately without add it to the registry
-func NewRuleState(rule *def.Rule) (rs *RuleState, err error) {
-	defer func() {
-		if err != nil {
-			if _, ok := err.(errorx.ErrorWithCode); !ok {
-				err = errorx.NewWithCode(errorx.RuleErr, err.Error())
-			}
-		}
-	}()
-
-	rs = &RuleState{
-		RuleId:   rule.Id,
-		Rule:     rule,
-		ActionCh: make(chan ActionSignal),
+// 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) *State {
+	contextLogger := conf.Log.WithField("Rule", rule.Id)
+	return &State{
+		Rule:         rule,
+		actionQ:      make([]ActionSignal, 0),
+		currentState: Stopped,
+		logger:       contextLogger,
 	}
-	err = infra.SafeRun(func() error {
-		if tp, err := planner.Plan(rule); err != nil {
+}
+
+func (s *State) WithTopo(topo *topo.Topo) *State {
+	s.topology = topo
+	return s
+}
+
+// Validate is the second level validation
+// It tries to plan and return any errors
+// Only run when creating the Rule
+func (s *State) Validate() error {
+	s.Lock()
+	defer s.Unlock()
+	err := infra.SafeRun(func() error {
+		if tp, err := planner.Plan(s.Rule); err != nil {
 			return err
 		} else {
-			rs.Topology = tp
-		}
-		if !rs.Rule.Triggered {
-			// manually force stop rule
-			rs.Topology.Cancel()
+			s.topology = tp
 		}
 		return nil
 	})
-	rs.run()
-	defer func() {
-		if err != nil {
-			rs.Close()
-		}
-	}()
-	return rs, err
+	return err
 }
 
-// UpdateTopo update the rule and the topology AND restart the topology
-// Do not need to call restart after update
-
-func (rs *RuleState) UpdateTopo(rule *def.Rule) (err error) {
-	defer func() {
-		if err != nil {
-			if _, ok := err.(errorx.ErrorWithCode); !ok {
-				err = errorx.NewWithCode(errorx.RuleErr, err.Error())
-			}
-		}
-	}()
-	t, err := planner.Plan(rule)
+func (s *State) transit(newState RunState, err error) {
+	s.Lock()
+	defer s.Unlock()
+	s.currentState = newState
 	if err != nil {
-		return err
-	}
-	if err := rs.Stop(); err != nil {
-		return err
+		s.lastWill = err.Error()
 	}
-	time.Sleep(1 * time.Millisecond)
-	rs.Rule = rule
-	rs.Topology = t
-	// If not triggered, just ignore start the rule
-	if rule.Triggered {
-		if err := rs.Start(); err != nil {
-			return err
-		}
+	switch newState {
+	case Running:
+		s.lastStartTimestamp = timex.GetNowInMilli()
+	case Stopped, StoppedByErr, ScheduledStop:
+		s.lastStopTimestamp = timex.GetNowInMilli()
+	default:
+		// do nothing
 	}
-	return nil
 }
 
-// only used for unit test
-var ignoreSignal = false
+func (s *State) GetState() RunState {
+	s.RLock()
+	defer s.RUnlock()
+	return s.currentState
+}
 
-// Run start to run the two loops, do not access any changeable states
-func (rs *RuleState) run() {
+// GetStatusMessage return the current RunState of the Rule
+// No set is provided, RunState are changed according to the action (start, stop)
+func (s *State) GetStatusMessage() string {
+	s.RLock()
+	defer s.RUnlock()
+	var result strings.Builder
+	result.WriteString("{")
+	// Compose status line
+	result.WriteString(`"status": "`)
+	result.WriteString(StateName[s.currentState])
+	result.WriteString(`",`)
+	result.WriteString(`"message": "`)
+	result.WriteString(s.lastWill)
+	result.WriteString(`",`)
+	// Compose run timing metrics
+	result.WriteString(`"lastStartTimestamp": "`)
+	result.WriteString(strconv.FormatInt(s.lastStartTimestamp, 10))
+	result.WriteString(`",`)
+	result.WriteString(`"lastStopTimestamp": "`)
+	result.WriteString(strconv.FormatInt(s.lastStopTimestamp, 10))
+	result.WriteString(`",`)
+	nextStartTimestamp := s.getNextScheduleStartTime()
+	result.WriteString(`"nextStartTimestamp": "`)
+	result.WriteString(strconv.FormatInt(nextStartTimestamp, 10))
+	result.WriteString(`",`)
+	// Compose metrics
 	var (
-		ctx    context.Context
-		cancel context.CancelFunc
+		keys   []string
+		values []any
 	)
-	// action loop, once start never end until the rule is deleted
-	go func() {
-		conf.Log.Infof("Start rulestate %s", rs.RuleId)
-		for {
-			s, opened := <-rs.ActionCh
-			if !opened {
-				conf.Log.Infof("Stop rulestate %s", rs.RuleId)
-				if cancel != nil {
-					cancel()
-				}
-				return
-			}
-
-			if ignoreSignal {
-				continue
-			}
-
-			switch s {
-			case ActionSignalStart:
-				if ctx != nil {
-					conf.Log.Warnf("rule %s is already started", rs.RuleId)
-				} else {
-					ctx, cancel = context.WithCancel(context.Background())
-					go rs.runTopo(ctx)
-				}
-			case ActionSignalStop:
-				// Stop the running loop
-				if cancel != nil {
-					cancel()
-					ctx = nil
-					cancel = nil
-				} else {
-					conf.Log.Warnf("rule %s is already stopped", rs.RuleId)
-				}
-			}
+	if s.topology != nil {
+		keys, values = s.topology.GetMetrics()
+	} else if len(s.stoppedMetrics) == 2 {
+		keys = s.stoppedMetrics[0].([]string)
+		values = s.stoppedMetrics[1].([]any)
+	}
+	if len(keys) > 0 {
+		for i, key := range keys {
+			value := values[i]
+			result.WriteString(`"`)
+			result.WriteString(key)
+			result.WriteString(`":`)
+			result.WriteString(`"`)
+			v, _ := cast.ToString(value, cast.CONVERT_ALL)
+			result.WriteString(v)
+			result.WriteString(`",`)
 		}
-	}()
-}
-
-func (rs *RuleState) runTopo(ctx context.Context) {
-	// Load the changeable states once
-	rs.Lock()
-	tp := rs.Topology
-	option := rs.Rule.Options.RestartStrategy
-	rs.Unlock()
-	if tp == nil {
-		conf.Log.Warnf("rule %s is not initialized or just stopped", rs.RuleId)
-		return
 	}
-	err := infra.SafeRun(func() error {
-		count := 0
-		d := time.Duration(option.Delay)
-		var er error
-		ticker := time.NewTicker(d)
-		defer ticker.Stop()
-		for {
-			select {
-			case e := <-tp.Open():
+	stStr := result.String()
+	stStr = stStr[:len(stStr)-1] + "}"
+	dst := &bytes.Buffer{}
+	var status string
+	if err := json.Indent(dst, cast.StringToBytes(stStr), "", "  "); err != nil {
+		status = stStr
+	} else {
+		status = dst.String()
+	}
+	return status
+}
 
-				er = e
-				if er != nil { // Only restart rule for errors
-					tp.GetContext().SetError(er)
-					conf.Log.Errorf("closing rule %s for error: %v", rs.RuleId, er)
-					tp.Cancel()
-				} else { // exit normally
-					return nil
-				}
-			}
-			if count < option.Attempts {
-				if d > time.Duration(option.MaxDelay) {
-					d = time.Duration(option.MaxDelay)
-				}
-				if option.JitterFactor > 0 {
-					d = time.Duration(math.Round(float64(d.Milliseconds())*((rand.Float64()*2-1)*option.JitterFactor+1))) * time.Millisecond
-					// make sure d is always in range
-					for d <= 0 || d > time.Duration(option.MaxDelay) {
-						d = time.Duration(math.Round(float64(d.Milliseconds())*((rand.Float64()*2-1)*option.JitterFactor+1))) * time.Millisecond
-					}
-					conf.Log.Infof("Rule %s will restart with jitterred delay %d", rs.RuleId, d)
-				} else {
-					conf.Log.Infof("Rule %s will restart with delay %d", rs.RuleId, d)
-				}
-				// retry after delay
-				select {
-				case <-ticker.C:
-					break
-				case <-ctx.Done():
-					conf.Log.Errorf("stop rule %s retry as cancelled", rs.RuleId)
-					return nil
-				}
-				count++
-				if option.Multiplier > 0 {
-					d = time.Duration(option.Delay) * time.Duration(math.Pow(option.Multiplier, float64(count)))
-				}
-			} else {
-				return er
-			}
-		}
-	})
-	if err != nil { // Exit after retries
-		rs.Lock()
-		// The only change the state by error
-		if rs.triggered != -1 {
-			rs.triggered = 0
-			if rs.Topology != nil {
-				rs.topoGraph = rs.Topology.GetTopo()
-			}
-			rs.ActionCh <- ActionSignalStop
+func (s *State) GetStatusMap() map[string]any {
+	s.RLock()
+	defer s.RUnlock()
+	result := make(map[string]any, 20)
+	result["status"] = StateName[s.currentState]
+	result["message"] = s.lastWill
+	result["lastStartTimestamp"] = s.lastStartTimestamp
+	result["lastStopTimestamp"] = s.lastStopTimestamp
+	nextStartTimestamp := s.getNextScheduleStartTime()
+	result["nextStartTimestamp"] = nextStartTimestamp
+	// Compose metrics
+	var (
+		keys   []string
+		values []any
+	)
+	if s.topology != nil {
+		keys, values = s.topology.GetMetrics()
+	} else if len(s.stoppedMetrics) == 2 {
+		keys = s.stoppedMetrics[0].([]string)
+		values = s.stoppedMetrics[1].([]any)
+	}
+	if len(keys) > 0 {
+		for i, key := range keys {
+			result[key] = values[i]
 		}
-
-		rs.Unlock()
 	}
+	return result
 }
 
-// The action functions are state machine.
-
-func (rs *RuleState) Start() (err error) {
-	defer func() {
-		if err != nil {
-			if _, ok := err.(errorx.ErrorWithCode); !ok {
-				err = errorx.NewWithCode(errorx.RuleErr, err.Error())
-			}
-		}
-	}()
-	rs.Lock()
-	defer rs.Unlock()
-	if rs.triggered == -1 {
-		return fmt.Errorf("rule %s is already deleted", rs.RuleId)
+// Start run start or add the start action to queue
+// 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 {
+	defer s.nextAction()
+	s.logger.Debug("start RunState")
+	done := s.triggerAction(ActionSignalStart)
+	if done {
+		return nil
 	}
-	if rs.Rule.IsLongRunningScheduleRule() {
-		isIn, err := schedule.IsInScheduleRanges(timex.GetNow(), rs.Rule.Options.CronDatetimeRange)
+	if s.Rule.IsLongRunningScheduleRule() {
+		isIn, err := schedule.IsInScheduleRanges(timex.GetNow(), s.Rule.Options.CronDatetimeRange)
 		if err != nil {
+			// s.transit(Stopped, err)
 			return err
 		}
-		// When rule is created, we need to check its schedule range before start it.
+		// When Rule is created, we need to check its schedule range before start it.
+		// If isIn, should be in running state, do not change it. And let it go to
 		if !isIn {
-			rs.triggered = 2
+			s.transit(ScheduledStop, nil)
 			return nil
 		}
-	}
-	if rs.Rule.IsScheduleRule() {
-		return rs.startScheduleRule()
-	}
-	return rs.start()
-}
-
-// startScheduleRule will register the job in the backgroundCron to run.
-// Job will do following 2 things:
-// 1. start the rule in cron if else the job is already stopped
-// 2. after the rule started, start an extract goroutine to stop the rule after specific duration
-func (rs *RuleState) startScheduleRule() error {
-	if rs.cronState.isInSchedule {
-		return fmt.Errorf("rule %s is already in schedule", rs.RuleId)
-	}
-	d, err := time.ParseDuration(rs.Rule.Options.Duration)
-	if err != nil {
-		return err
-	}
-	var cronCtx context.Context
-	cronCtx, rs.cronState.cancel = context.WithCancel(context.Background())
-	now := timex.GetNow()
-	isInRunningSchedule, remainedDuration, err := rs.isInRunningSchedule(now, d)
-	if err != nil {
-		return err
-	}
-	if isInRunningSchedule {
-		if err := rs.runScheduleRule(); err != nil {
+	} else if s.Rule.IsScheduleRule() {
+		err := s.registerScheduleRule()
+		if err != nil {
 			return err
+		} else {
+			s.transit(ScheduledStop, nil)
+			return nil
 		}
-		rs.stopAfterDuration(remainedDuration, cronCtx)
 	}
-	entryID, err := backgroundCron.AddFunc(rs.Rule.Options.Cron, func() {
-		var started bool
-		var err error
-		if started, err = func() (bool, error) {
-			switch backgroundCron.(type) {
-			case *MockCron:
-				// skip mutex if this is a unit test
-			default:
-				rs.Lock()
-				defer rs.Unlock()
-			}
-			now := timex.GetNow()
-			allowed, err := rs.isInAllowedTimeRange(now)
-			if err != nil {
-				return false, err
-			}
-			if !allowed {
-				return false, nil
-			}
-
-			rs.cronState.cron = rs.Rule.Options.Cron
-			rs.cronState.duration = rs.Rule.Options.Duration
-			return true, rs.start()
-		}(); err != nil {
-			rs.Lock()
-			rs.cronState.startFailedCnt++
-			rs.Unlock()
-			conf.Log.Errorf(err.Error())
-			return
-		}
-		if started {
-			rs.stopAfterDuration(d, cronCtx)
-		}
-	})
+	// Start normally or start in schedule period Rule
+	// doStart trigger the Rule run. If no trigger error, the Rule will run async and control the state by itself
+	err := s.doStart()
 	if err != nil {
+		s.transit(StoppedByErr, err)
 		return err
+	} else {
+		s.transit(Running, nil)
 	}
-	rs.cronState.isInSchedule = true
-	rs.cronState.entryID = entryID
 	return nil
 }
 
-func (rs *RuleState) runScheduleRule() error {
-	rs.cronState.cron = rs.Rule.Options.Cron
-	rs.cronState.duration = rs.Rule.Options.Duration
-	err := rs.start()
+func (s *State) ScheduleStart() error {
+	defer s.nextAction()
+	s.logger.Debug("scheduled start RunState")
+	done := s.triggerAction(ActionSignalScheduledStart)
+	if done {
+		return nil
+	}
+	// doStart trigger the Rule run. If no trigger error, the Rule will run async and control the state by itself
+	err := s.doStart()
 	if err != nil {
+		s.transit(StoppedByErr, err)
 		return err
+	} else {
+		s.transit(Running, nil)
 	}
 	return nil
 }
 
-// stopAfterDuration only for schedule rule
-func (rs *RuleState) stopAfterDuration(d time.Duration, cronCtx context.Context) {
-	after := time.After(d)
-	go func(ctx context.Context) {
-		select {
-		case <-after:
-			rs.Lock()
-			defer rs.Unlock()
-			if err := rs.internalStop(); err != nil {
-				conf.Log.Errorf("close rule %s failed, err: %v", rs.RuleId, err)
-			}
-			return
-		case <-cronCtx.Done():
-			return
-		}
-	}(cronCtx)
-}
-
-func (rs *RuleState) start() error {
-	if rs.triggered != 1 {
-		// If the rule has been stopped due to error, the topology is not nil
-		if rs.Topology != nil {
-			if rs.Topology.HasOpen() {
-				rs.Topology.Cancel()
-				if tp, err := planner.Plan(rs.Rule); err != nil {
-					return err
-				} else {
-					rs.Topology = tp
-				}
-			}
+func (s *State) triggerAction(action ActionSignal) bool {
+	s.Lock()
+	defer s.Unlock()
+	if len(s.actionQ) > 0 {
+		if s.actionQ[len(s.actionQ)-1] == action {
+			s.logger.Infof("ignore action %d because last action is the same", action)
+			return true
+		} else {
+			s.actionQ = append(s.actionQ, action)
+			s.logger.Infof("defer action %d to action queue", action)
+			return true
 		}
-		rs.triggered = 1
 	}
-	if rs.Rule.IsScheduleRule() || rs.Rule.IsLongRunningScheduleRule() {
-		conf.Log.Debugf("rule %v started", rs.RuleId)
+	ss := s.currentState
+	switch action {
+	case ActionSignalStart:
+		switch ss {
+		case Starting, Running, ScheduledStop:
+			s.logger.Infof("ignore start action, because current RunState is %s", StateName[ss])
+			return true
+		case Stopping:
+			s.actionQ = append(s.actionQ, ActionSignalStart)
+			s.logger.Infof("defer start action to action queue because current RunState is stopping")
+			return true
+		case Stopped, StoppedByErr:
+			s.currentState = Starting
+			return false
+		}
+	case ActionSignalStop:
+		switch ss {
+		case Stopped, StoppedByErr:
+			s.logger.Infof("ignore stop action, because current RunState is %s", StateName[ss])
+			return true
+		case Starting, Stopping:
+			s.actionQ = append(s.actionQ, action)
+			s.logger.Infof("defer stop action to action queue because current RunState is starting")
+			return true
+		case Running, ScheduledStop: // do stop
+			s.currentState = Stopping
+			return false
+		}
+	case ActionSignalScheduledStart:
+		switch ss {
+		case ScheduledStop, Stopped, StoppedByErr:
+			s.currentState = Starting
+			return false
+		case Starting, Running:
+			s.logger.Infof("ignore schedule start action, because current RunState is %s", StateName[ss])
+			return true
+		case Stopping:
+			s.actionQ = append(s.actionQ, action)
+			s.logger.Infof("defer schedule start action to action queue because current RunState is stopping")
+			return true
+		}
+	case ActionSignalScheduledStop:
+		switch ss {
+		case Running:
+			s.currentState = Stopping
+			return false
+		case ScheduledStop, Stopped, StoppedByErr:
+			s.logger.Infof("ignore schedule stop action, because current RunState is %s", StateName[ss])
+			return true
+		case Starting, Stopping:
+			s.actionQ = append(s.actionQ, action)
+			s.logger.Infof("defer schedule stop action to action queue because current RunState is %s", StateName[ss])
+			return true
+		}
 	}
-	rs.lastStartTimestamp = time.Now().UnixMilli()
-	rs.ActionCh <- ActionSignalStart
-	return nil
+	return false
 }
 
-// Stop remove the Topology
-func (rs *RuleState) Stop() (err error) {
-	defer func() {
-		if err != nil {
-			if _, ok := err.(errorx.ErrorWithCode); !ok {
-				err = errorx.NewWithCode(errorx.RuleErr, err.Error())
-			}
-		}
-	}()
-	rs.Lock()
-	defer rs.Unlock()
-	if rs.Rule.IsScheduleRule() || rs.Rule.IsLongRunningScheduleRule() {
-		conf.Log.Debugf("rule %v manual stopped", rs.RuleId)
+// Stop run stop action or add the stop action to queue
+// regSchedule: whether need to handle scheduler. If call externally, set it to true
+func (s *State) Stop() {
+	defer s.nextAction()
+	s.logger.Debug("stop RunState")
+	done := s.triggerAction(ActionSignalStop)
+	if done {
+		return
 	}
-	rs.stopScheduleRule()
-	return rs.stop()
+	// unregister if it is the scheduled Rule
+	s.removeScheduleRule()
+	// do stop, stopping action and starting action are mutual exclusive. No concurrent problem here
+	err := s.doStop()
+	if err == nil {
+		err = errors.New("canceled manually")
+	}
+	// currentState may be accessed concurrently
+	s.transit(Stopped, err)
+	return
 }
 
-func (rs *RuleState) stopScheduleRule() {
-	if rs.Rule.IsScheduleRule() && rs.cronState.isInSchedule {
-		rs.cronState.isInSchedule = false
-		if rs.cronState.cancel != nil {
-			rs.cronState.cancel()
-		}
-		rs.cronState.startFailedCnt = 0
-		backgroundCron.Remove(rs.cronState.entryID)
+func (s *State) ScheduleStop() {
+	defer s.nextAction()
+	s.logger.Debug("scheduled stop RunState")
+	done := s.triggerAction(ActionSignalScheduledStop)
+	if done {
+		return
+	}
+	// do stop, stopping action and starting action are mutual exclusive. No concurrent problem here
+	err := s.doStop()
+	// currentState may be accessed concurrently
+	if schedule.IsAfterTimeRanges(timex.GetNow(), s.Rule.Options.CronDatetimeRange) {
+		s.transit(ScheduledStop, errors.New("schedule terminated"))
+	} else {
+		s.transit(ScheduledStop, err)
 	}
+	return
 }
 
-func (rs *RuleState) stop() error {
-	if rs.triggered == -1 {
-		return fmt.Errorf("rule %s is already deleted", rs.RuleId)
+func (s *State) nextAction() {
+	var action ActionSignal = -1
+	s.Lock()
+	if len(s.actionQ) > 0 {
+		action = s.actionQ[0]
+		s.actionQ = s.actionQ[1:]
+	}
+	s.Unlock()
+	var err error
+	switch action {
+	case ActionSignalStart:
+		err = s.Start()
+	case ActionSignalStop:
+		s.Stop()
+	case ActionSignalScheduledStart:
+		err = s.ScheduleStart()
+	case ActionSignalScheduledStop:
+		s.ScheduleStop()
 	}
-	rs.triggered = 0
-	if rs.Topology != nil {
-		rs.Topology.Cancel()
-		// wait all operator close
-		rs.Topology.WaitClose()
-		// de-reference old Topology in order to release data memory
-		rs.Topology = rs.Topology.NewTopoWithSucceededCtx()
+	if err != nil {
+		s.logger.Error(err)
 	}
-	rs.lastStopTimestamp = time.Now().UnixMilli()
-	rs.ActionCh <- ActionSignalStop
-	return nil
 }
 
-func (rs *RuleState) InternalStop() (err error) {
-	defer func() {
-		if err != nil {
-			if _, ok := err.(errorx.ErrorWithCode); !ok {
-				err = errorx.NewWithCode(errorx.RuleErr, err.Error())
+// doStart/doStop actions are run in sync!!
+// 1. create topo if not exists
+// 2. run topo async
+func (s *State) doStart() error {
+	err := infra.SafeRun(func() error {
+		if s.topology == nil {
+			if tp, err := planner.Plan(s.Rule); err != nil {
+				return err
+			} else {
+				s.topology = tp
 			}
 		}
-	}()
-	rs.Lock()
-	defer rs.Unlock()
-	if !rs.Rule.IsLongRunningScheduleRule() {
-		err := fmt.Errorf("rule %v isn't allowed to execute Internal stop as it's not long running schedule rule", rs.RuleId)
-		conf.Log.Errorf(err.Error())
-		return err
-	} else {
-		conf.Log.Debugf("rule %v internal stopped", rs.RuleId)
-	}
-	return rs.internalStop()
+		ctx, cancel := context.WithCancel(context.Background())
+		s.cancelRetry = cancel
+		s.lastStartTimestamp = timex.GetNowInMilli()
+		s.lastWill = ""
+		go s.runTopo(ctx, s.topology, s.Rule.Options.RestartStrategy)
+		return nil
+	})
+	return err
 }
 
-func (rs *RuleState) internalStop() error {
-	if rs.triggered == -1 {
-		return fmt.Errorf("rule %s is already deleted", rs.RuleId)
+func (s *State) doStop() error {
+	if s.cancelRetry != nil {
+		s.cancelRetry()
 	}
-	rs.triggered = 2
-	if rs.Topology != nil {
-		rs.Topology.Cancel()
-		// de-reference old Topology in order to release data memory
-		rs.Topology = rs.Topology.NewTopoWithSucceededCtx()
+	if s.topology != nil {
+		e := s.topology.GetContext().Err()
+		s.topology.Cancel()
+		s.topology.WaitClose()
+		s.topoGraph = s.topology.GetTopo()
+		keys, values := s.topology.GetMetrics()
+		s.stoppedMetrics = []any{keys, values}
+		s.topology = nil
+		return e
 	}
-	rs.ActionCh <- ActionSignalStop
 	return nil
 }
 
-func (rs *RuleState) Close() (err error) {
-	defer func() {
-		if err != nil {
-			if _, ok := err.(errorx.ErrorWithCode); !ok {
-				err = errorx.NewWithCode(errorx.RuleErr, err.Error())
+// 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 er != nil { // Only restart Rule for errors
+					tp.GetContext().SetError(er)
+					s.logger.Errorf("closing Rule for error: %v", er)
+					tp.Cancel()
+				} else { // exit normally
+					return nil
+				}
+				// Although it is stopped, it is still retrying, so the status is still RUNNING
+				s.lastWill = "retrying after error: " + er.Error()
+			}
+			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
 			}
 		}
-	}()
-	rs.Lock()
-	defer rs.Unlock()
-	if rs.isClosed {
-		return nil
-	}
-	if rs.Topology != nil {
-		rs.Topology.RemoveMetrics()
+	})
+	if err != nil { // Exit after retries
+		s.logger.Error(err)
+		if s.topology != nil {
+			s.topoGraph = s.topology.GetTopo()
+			keys, values := s.topology.GetMetrics()
+			s.stoppedMetrics = []any{keys, values}
+			s.topology = nil
+		}
+		s.transit(StoppedByErr, err)
+	} else {
+		s.transit(Stopped, err)
 	}
-	if rs.triggered == 1 && rs.Topology != nil {
-		rs.Topology.Cancel()
-		rs.Topology.WaitClose()
+}
+
+// Other APIs
+
+func (s *State) GetTopoGraph() *def.PrintableTopo {
+	s.RLock()
+	defer s.RUnlock()
+	if s.topology != nil {
+		return s.topology.GetTopo()
+	} else {
+		return s.topoGraph
 	}
-	rs.triggered = -1
-	rs.stopScheduleRule()
-	close(rs.ActionCh)
-	rs.isClosed = true
-	return nil
 }
 
-func (rs *RuleState) GetState() (s string, err error) {
+func (s *State) Delete() (err error) {
 	defer func() {
 		if err != nil {
 			if _, ok := err.(errorx.ErrorWithCode); !ok {
@@ -561,95 +561,46 @@ func (rs *RuleState) GetState() (s string, err error) {
 			}
 		}
 	}()
-	rs.RLock()
-	defer rs.RUnlock()
-	result := ""
-	if rs.Topology == nil {
-		result = "Stopped: fail to create the topo."
-	} else {
-		c := (*rs.Topology).GetContext()
-		if c != nil {
-			err := c.Err()
-			switch err {
-			case nil:
-				result = RuleStarted
-			case context.Canceled:
-				result = rs.getStoppedRuleState()
-			case context.DeadlineExceeded:
-				result = "Stopped: deadline exceed."
-			default:
-				result = fmt.Sprintf("Stopped: %v.", err)
-			}
-		} else {
-			result = rs.getStoppedRuleState()
-		}
-	}
-	if rs.Rule.IsScheduleRule() && rs.cronState.startFailedCnt > 0 {
-		result = result + fmt.Sprintf(" Start failed count: %v.", rs.cronState.startFailedCnt)
-	}
-	return result, nil
-}
-
-func (rs *RuleState) getStoppedRuleState() (result string) {
-	if schedule.IsAfterTimeRanges(timex.GetNow(), rs.Rule.Options.CronDatetimeRange) {
-		result = RuleTerminated
-	} else if rs.cronState.isInSchedule {
-		result = RuleWait
-	} else if rs.triggered == 0 || rs.triggered == -1 {
-		result = RuleStopped
-	} else if rs.triggered == 2 {
-		result = RuleWait
-	}
-	return result
+	s.Lock()
+	defer s.Unlock()
+	if s.topology != nil {
+		s.topology.RemoveMetrics()
+		s.topology.Cancel()
+		s.topology.WaitClose()
+	}
+	s.removeScheduleRule()
+	return nil
 }
 
-func (rs *RuleState) GetTopoGraph() *def.PrintableTopo {
-	rs.RLock()
-	defer rs.RUnlock()
-	if rs.topoGraph != nil {
-		return rs.topoGraph
-	} else if rs.Topology != nil {
-		return rs.Topology.GetTopo()
-	} else {
-		return nil
+func (s *State) GetMetrics() ([]string, []any) {
+	s.RLock()
+	defer s.RUnlock()
+	if s.topology != nil {
+		return s.topology.GetMetrics()
 	}
+	return nil, nil
 }
 
-func (rs *RuleState) isInRunningSchedule(now time.Time, d time.Duration) (bool, time.Duration, error) {
-	allowed, err := rs.isInAllowedTimeRange(now)
-	if err != nil {
-		return false, 0, err
+func (s *State) GetStreams() []string {
+	s.RLock()
+	defer s.RUnlock()
+	if s.topology != nil {
+		return s.topology.GetStreams()
 	}
-	if !allowed {
-		return false, 0, nil
-	}
-	cronExpr := rs.Rule.Options.Cron
-	if strings.HasPrefix(cronExpr, "mock") {
-		return false, 0, nil
-	}
-	return schedule.IsInRunningSchedule(cronExpr, now, d)
+	return nil
 }
 
-func (rs *RuleState) isInAllowedTimeRange(now time.Time) (bool, error) {
-	return schedule.IsInScheduleRanges(now, rs.Rule.Options.CronDatetimeRange)
+func (s *State) GetLastWill() string {
+	s.RLock()
+	defer s.RUnlock()
+	return s.lastWill
 }
 
-func (rs *RuleState) GetNextScheduleStartTime() int64 {
-	if rs.Rule.IsScheduleRule() && len(rs.Rule.Options.Cron) > 0 {
-		isIn, err := schedule.IsInScheduleRanges(time.Now(), rs.Rule.Options.CronDatetimeRange)
-		if err == nil && isIn {
-			s, err := cron.ParseStandard(rs.Rule.Options.Cron)
-			if err == nil {
-				return s.Next(time.Now()).UnixMilli()
-			}
-		}
+func (s *State) ResetStreamOffset(name string, input map[string]any) error {
+	s.RLock()
+	defer s.RUnlock()
+	if s.topology != nil {
+		return s.topology.ResetStreamOffset(name, input)
 	}
-	return 0
-}
-
-func (rs *RuleState) GetScheduleTimestamp() (int64, int64, int64) {
-	nextStartTimestamp := rs.GetNextScheduleStartTime()
-	rs.Lock()
-	defer rs.Unlock()
-	return rs.lastStartTimestamp, rs.lastStopTimestamp, nextStartTimestamp
+	return fmt.Errorf("topo is not initialized, check rule status")
 }
diff --git a/internal/topo/topo.go b/internal/topo/topo.go
index 8f96d7fb..dbdb21fd 100644
--- a/internal/topo/topo.go
+++ b/internal/topo/topo.go
@@ -97,13 +97,6 @@ func (s *Topo) GetContext() api.StreamContext {
 	return s.ctx
 }
 
-func (s *Topo) NewTopoWithSucceededCtx() *Topo {
-	n := &Topo{}
-	n.ctx = s.ctx
-	n.cancel = s.cancel
-	return n
-}
-
 func (s *Topo) GetName() string {
 	return s.name
 }


Test patch:
diff --git a/internal/server/rest_test.go b/internal/server/rest_test.go
index 99797291..98b2dfc2 100644
--- a/internal/server/rest_test.go
+++ b/internal/server/rest_test.go
@@ -35,7 +35,6 @@ import (
 
 	"github.com/lf-edge/ekuiper/v2/internal/conf"
 	"github.com/lf-edge/ekuiper/v2/internal/io/http/httpserver"
-	"github.com/lf-edge/ekuiper/v2/internal/meta"
 	"github.com/lf-edge/ekuiper/v2/internal/pkg/def"
 	"github.com/lf-edge/ekuiper/v2/internal/pkg/store"
 	"github.com/lf-edge/ekuiper/v2/internal/processor"
@@ -51,7 +50,7 @@ func init() {
 	streamProcessor = processor.NewStreamProcessor()
 	ruleProcessor = processor.NewRuleProcessor()
 	rulesetProcessor = processor.NewRulesetProcessor(ruleProcessor, streamProcessor)
-	registry = &RuleRegistry{internal: make(map[string]*rule.RuleState)}
+	registry = &RuleRegistry{internal: make(map[string]*rule.State)}
 	uploadsDb, _ = store.GetKV("uploads")
 	uploadsStatusDb, _ = store.GetKV("uploadsStatusDb")
 	sysMetrics = NewMetrics()
@@ -248,7 +247,7 @@ func (suite *RestTestSuite) TestRecoverRule() {
 	for _, s := range got {
 		if s["id"] == "recoverTest" {
 			find = true
-			require.Equal(suite.T(), "Stopped: canceled manually.", s["status"])
+			require.Equal(suite.T(), "Stopped", s["status"])
 		}
 	}
 	require.True(suite.T(), find)
@@ -288,7 +287,7 @@ func (suite *RestTestSuite) Test_rulesManageHandler() {
 				continue
 			}
 			// err = server.StartRule(rule, &reply)
-			reply = recoverRule(rule)
+			reply = registry.RecoverRule(rule)
 			if 0 != len(reply) {
 				logger.Info(reply)
 			}
@@ -332,7 +331,6 @@ func (suite *RestTestSuite) Test_rulesManageHandler() {
 
 	// validate a rule
 	ruleJson := `{"id": "rule1","triggered": false,"sql": "select * from alert","actions": [{"log": {}}]}`
-
 	buf2 := bytes.NewBuffer([]byte(ruleJson))
 	req2, _ := http.NewRequest(http.MethodPost, "http://localhost:8080/rules/validate", buf2)
 	w2 := httptest.NewRecorder()
@@ -344,7 +342,6 @@ func (suite *RestTestSuite) Test_rulesManageHandler() {
 
 	// validate a wrong rule
 	ruleJson = `{"id": "rule321", "sql": "select * from alert"}`
-
 	buf2 = bytes.NewBuffer([]byte(ruleJson))
 	req2, _ = http.NewRequest(http.MethodPost, "http://localhost:8080/rules/validate", buf2)
 	w2 = httptest.NewRecorder()
@@ -361,7 +358,6 @@ func (suite *RestTestSuite) Test_rulesManageHandler() {
 
 	// create rule with trigger false
 	ruleJson = `{"id": "rule3/21","triggered": false,"sql": "select * from alert","actions": [{"log": {}}]}`
-
 	buf2 = bytes.NewBuffer([]byte(ruleJson))
 	req2, _ = http.NewRequest(http.MethodPost, "http://localhost:8080/rules", buf2)
 	w2 = httptest.NewRecorder()
@@ -373,7 +369,6 @@ func (suite *RestTestSuite) Test_rulesManageHandler() {
 
 	// create rule with trigger false
 	ruleJson = `{"id": "rule321","triggered": false,"sql": "select * from alert","actions": [{"log": {}}]}`
-
 	buf2 = bytes.NewBuffer([]byte(ruleJson))
 	req2, _ = http.NewRequest(http.MethodPost, "http://localhost:8080/rules", buf2)
 	w2 = httptest.NewRecorder()
@@ -383,34 +378,28 @@ func (suite *RestTestSuite) Test_rulesManageHandler() {
 	req3, _ := http.NewRequest(http.MethodGet, "http://localhost:8080/rules", bytes.NewBufferString("any"))
 	w3 := httptest.NewRecorder()
 	suite.r.ServeHTTP(w3, req3)
-
 	_, _ = io.ReadAll(w3.Result().Body)
 
 	// update rule, will set rule to triggered
 	ruleJson = `{"id": "rule321","triggered": true,"sql": "select * from alert","actions": [{"nop": {}}]}`
-
 	buf2 = bytes.NewBuffer([]byte(ruleJson))
 	req1, _ = http.NewRequest(http.MethodPut, "http://localhost:8080/rules/rule321", buf2)
 	w1 = httptest.NewRecorder()
 	suite.r.ServeHTTP(w1, req1)
-
 	assert.Equal(suite.T(), http.StatusOK, w1.Code)
 
 	// update wrong rule
 	ruleJson = `{"id": "rule321","sql": "select * from alert1","actions": [{"nop": {}}]}`
-
 	buf2 = bytes.NewBuffer([]byte(ruleJson))
 	req1, _ = http.NewRequest(http.MethodPut, "http://localhost:8080/rules/rule321", buf2)
 	w1 = httptest.NewRecorder()
 	suite.r.ServeHTTP(w1, req1)
-
 	assert.Equal(suite.T(), http.StatusBadRequest, w1.Code)
 
 	// get rule
 	req1, _ = http.NewRequest(http.MethodGet, "http://localhost:8080/rules/rule321", bytes.NewBufferString("any"))
 	w1 = httptest.NewRecorder()
 	suite.r.ServeHTTP(w1, req1)
-
 	returnVal, _ = io.ReadAll(w1.Result().Body)
 	expect = `{"id": "rule321","triggered": true,"sql": "select * from alert","actions": [{"nop": {}}]}`
 	assert.Equal(suite.T(), expect, string(returnVal))
@@ -453,7 +442,6 @@ func (suite *RestTestSuite) Test_rulesManageHandler() {
 	w1 = httptest.NewRecorder()
 	suite.r.ServeHTTP(w1, req1)
 	returnVal, _ = io.ReadAll(w1.Result().Body)
-
 	expect = "{\"sources\":[\"source_alert\"],\"edges\":{\"op_2_decoder\":[\"op_3_project\"],\"op_3_project\":[\"op_nop_0_0_transform\"],\"op_nop_0_0_transform\":[\"op_nop_0_1_encode\"],\"op_nop_0_1_encode\":[\"sink_nop_0\"],\"source_alert\":[\"op_2_decoder\"]}}"
 	assert.Equal(suite.T(), expect, string(returnVal))
 
@@ -491,7 +479,6 @@ func (suite *RestTestSuite) Test_rulesManageHandler() {
 	equal, err = assertErrorCode(errorx.NOT_FOUND, returnVal)
 	require.NoError(suite.T(), err)
 	require.True(suite.T(), equal)
-
 	assert.Equal(suite.T(), http.StatusNotFound, w1.Code)
 
 	// update rule, will set rule to triggered
@@ -507,7 +494,6 @@ func (suite *RestTestSuite) Test_rulesManageHandler() {
 	w1 = httptest.NewRecorder()
 	suite.r.ServeHTTP(w1, req1)
 	returnVal, _ = io.ReadAll(w1.Result().Body)
-
 	expect = `Rule rule321 was restarted`
 	assert.Equal(suite.T(), expect, string(returnVal))
 
@@ -515,7 +501,6 @@ func (suite *RestTestSuite) Test_rulesManageHandler() {
 	req1, _ = http.NewRequest(http.MethodGet, "http://localhost:8080/rules/rule321", bytes.NewBufferString("any"))
 	w1 = httptest.NewRecorder()
 	suite.r.ServeHTTP(w1, req1)
-
 	returnVal, _ = io.ReadAll(w1.Result().Body)
 	expect = `{"triggered":true,"id":"rule321","sql":"select * from alert","actions":[{"nop":{}}],"options":{"lateTolerance":"1s","concurrency":1,"bufferLength":1024,"sendError":true,"checkpointInterval":"5m0s","restartStrategy":{"delay":"1s","multiplier":2,"maxDelay":"30s","jitterFactor":0.1}}}`
 	assert.Equal(suite.T(), expect, string(returnVal))
@@ -903,35 +888,6 @@ func (suite *RestTestSuite) TestUpdateRuleOffset() {
 	require.Equal(suite.T(), `success`, returnStr)
 }
 
-func (suite *RestTestSuite) TestCreateRuleReplacePasswd() {
-	meta.InitYamlConfigManager()
-	confKeyJson := `{"insecureSkipVerify":false,"protocolVersion":"3.1.1","qos":1,"server":"tcp://122.9.166.75:1883","token":"123","password":"4444"}`
-	req, _ := http.NewRequest(http.MethodPut, "http://localhost:8080/metadata/sinks/mqtt/confKeys/broker", bytes.NewBufferString(confKeyJson))
-	w := httptest.NewRecorder()
-	suite.r.ServeHTTP(w, req)
-	require.Equal(suite.T(), http.StatusOK, w.Code)
-
-	buf1 := bytes.NewBuffer([]byte(`{"sql":"CREATE stream demodemo() WITH (DATASOURCE=\"0\", TYPE=\"mqtt\")"}`))
-	req1, _ := http.NewRequest(http.MethodPost, "http://localhost:8080/streams", buf1)
-	w1 := httptest.NewRecorder()
-	suite.r.ServeHTTP(w1, req1)
-
-	ruleJson2 := `{"id":"test1234","triggered":false,"sql":"select * from demodemo","actions":[{"mqtt":{"password":"******","topic":"123","server":"123","resourceId":"broker"}}]}`
-	buf2 := bytes.NewBuffer([]byte(ruleJson2))
-	req2, _ := http.NewRequest(http.MethodPost, "http://localhost:8080/rules", buf2)
-	w2 := httptest.NewRecorder()
-	suite.r.ServeHTTP(w2, req2)
-	require.Equal(suite.T(), http.StatusCreated, w2.Code)
-
-	r, err := ruleProcessor.GetRuleById("test1234")
-	require.NoError(suite.T(), err)
-	mqttSinkConfig := r.Actions[0]["mqtt"]
-	require.NotNil(suite.T(), mqttSinkConfig)
-	c, ok := mqttSinkConfig.(map[string]interface{})
-	require.True(suite.T(), ok)
-	require.Equal(suite.T(), "4444", c["password"])
-}
-
 func (suite *RestTestSuite) TestCreateDuplicateRule() {
 	buf1 := bytes.NewBuffer([]byte(`{"sql":"CREATE stream demo123() WITH (DATASOURCE=\"0\", TYPE=\"mqtt\")"}`))
 	req1, _ := http.NewRequest(http.MethodPost, "http://localhost:8080/streams", buf1)
@@ -989,34 +945,6 @@ func (suite *RestTestSuite) TestGetAllRuleStatus() {
 	require.True(suite.T(), ok)
 }
 
-func (suite *RestTestSuite) TestSinkHiddenPassword() {
-	buf1 := bytes.NewBuffer([]byte(`{"sql":"CREATE stream demo78() WITH (DATASOURCE=\"0\", TYPE=\"mqtt\")"}`))
-	req1, _ := http.NewRequest(http.MethodPost, "http://localhost:8080/streams", buf1)
-	w1 := httptest.NewRecorder()
-	suite.r.ServeHTTP(w1, req1)
-
-	ruleJson2 := `{"triggered":false,"id":"rule34","sql":"select * from demo78;","actions":[{"mqtt":{"server":"tcp://broker.emqx.io:1883","topic":"devices/demo_001/messages/events/","qos":0,"clientId":"demo_001","username":"xyz.azure-devices.net/demo_001/?api-version=2018-06-30","password":"12345"}}]}`
-	buf2 := bytes.NewBuffer([]byte(ruleJson2))
-	req2, _ := http.NewRequest(http.MethodPost, "http://localhost:8080/rules", buf2)
-	w2 := httptest.NewRecorder()
-	suite.r.ServeHTTP(w2, req2)
-	require.Equal(suite.T(), http.StatusCreated, w2.Code)
-
-	ruleJson2 = `{"triggered":false,"id":"rule34","sql":"select * from demo78;","actions":[{"mqtt":{"server":"tcp://broker.emqx.io:1883","topic":"devices/demo_001/messages/events/","qos":0,"clientId":"demo_001","username":"xyz.azure-devices.net/demo_001/?api-version=2018-06-30","password":"******"}}]}`
-	buf2 = bytes.NewBuffer([]byte(ruleJson2))
-	req2, _ = http.NewRequest(http.MethodPut, "http://localhost:8080/rules/rule34", buf2)
-	w2 = httptest.NewRecorder()
-	suite.r.ServeHTTP(w2, req2)
-	require.Equal(suite.T(), http.StatusOK, w2.Code)
-
-	ruleJson, err := ruleProcessor.GetRuleJson("rule34")
-	require.NoError(suite.T(), err)
-	r := &def.Rule{}
-	require.NoError(suite.T(), json.Unmarshal([]byte(ruleJson), r))
-	m := r.Actions[0]["mqtt"].(map[string]interface{})
-	require.Equal(suite.T(), "12345", m["password"])
-}
-
 func (suite *RestTestSuite) TestWaitStopRule() {
 	ip := "127.0.0.1"
 	port := 10085
diff --git a/internal/server/rpc_test.go b/internal/server/rpc_test.go
index d196db4b..07da6ead 100644
--- a/internal/server/rpc_test.go
+++ b/internal/server/rpc_test.go
@@ -15,14 +15,13 @@
 package server
 
 import (
-	"encoding/json"
 	"fmt"
 	"os"
+	"strings"
 	"testing"
 	"time"
 
 	"github.com/stretchr/testify/assert"
-	"github.com/stretchr/testify/require"
 	"github.com/stretchr/testify/suite"
 
 	"github.com/lf-edge/ekuiper/v2/internal/meta"
@@ -65,7 +64,7 @@ func (suite *ServerTestSuite) TestStream() {
 	assert.Nil(suite.T(), err)
 	assert.Equal(suite.T(), "Query was submit successfully.", reply)
 
-	var result string = ""
+	var result string
 	for i := 0; i < 5; i++ {
 		var queryresult string
 		time.Sleep(time.Second)
@@ -76,7 +75,9 @@ func (suite *ServerTestSuite) TestStream() {
 			break
 		}
 	}
-	assert.Equal(suite.T(), "[{\"humidity\":50,\"temperature\":22.5}]", result)
+	allResults := strings.Split(result, "\n")
+	assert.True(suite.T(), len(allResults) >= 1)
+	assert.Equal(suite.T(), "[{\"humidity\":50,\"temperature\":22.5}]", allResults[0])
 	stopQuery()
 }
 
@@ -225,14 +226,3 @@ func (suite *ServerTestSuite) TearDownTest() {
 func TestServerTestSuite(t *testing.T) {
 	suite.Run(t, new(ServerTestSuite))
 }
-
-func TestGetStoppedMessage(t *testing.T) {
-	message := `"123","123","123"`
-	r, err := getStoppedState(message)
-	require.NoError(t, err)
-	v := map[string]string{}
-	err = json.Unmarshal([]byte(r), &v)
-	require.NoError(t, err)
-	require.Equal(t, "stopped", v["status"])
-	require.Equal(t, message, v["message"])
-}
diff --git a/internal/server/rule_manager_test.go b/internal/server/rule_manager_test.go
new file mode 100644
index 00000000..48ef1fd6
--- /dev/null
+++ b/internal/server/rule_manager_test.go
@@ -0,0 +1,55 @@
+// Copyright 2024 EMQ Technologies Co., Ltd.
+//
+// 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 server
+
+import (
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+
+	"github.com/lf-edge/ekuiper/v2/internal/pkg/def"
+	"github.com/lf-edge/ekuiper/v2/internal/topo/rule"
+)
+
+func TestErrors(t *testing.T) {
+	// update invalid rule
+	err := registry.UpdateRule("test", "selectabc")
+	assert.EqualError(t, err, "Invalid rule json: Parse rule selectabc error : invalid character 's' looking for beginning of value.")
+	// update rule, no id
+	err = registry.UpdateRule("test", `{"id":"test","sql":"SELECT * FROM demo","actions":[{"log":{}}]}`)
+	assert.EqualError(t, err, "Rule test is not found in registry, please check if it is created")
+	// delete rule, no id
+	err = registry.DeleteRule("test")
+	assert.EqualError(t, err, "rule test not found")
+	// restart rule, no id
+	err = registry.RestartRule("test")
+	assert.EqualError(t, err, "Rule test is not found in registry, please check if it is created")
+	_, err = registry.GetRuleStatus("test")
+	assert.EqualError(t, err, "Rule test is not found")
+	_, err = registry.GetRuleStatusV2("test")
+	assert.EqualError(t, err, "Rule test is not found")
+	_, err = registry.GetRuleTopo("test")
+	assert.EqualError(t, err, "Rule test is not found")
+	err = registry.scheduledStart("test")
+	assert.EqualError(t, err, "Scheduled rule test is not found in registry, please check if it is deleted")
+	err = registry.scheduledStop("test")
+	assert.EqualError(t, err, "Scheduled rule test is not found in registry, please check if it is deleted")
+	err = registry.stopAtExit("test")
+	assert.EqualError(t, err, "Rule test is not found in registry, please check if it is deleted")
+	// db problems
+	registry.register("test", rule.NewState(def.GetDefaultRule("testErrors", "select * from demo")))
+	err = registry.StartRule("test")
+	assert.EqualError(t, err, "fail to get stream demo, please check if stream is created")
+}
diff --git a/internal/server/server_test.go b/internal/server/server_test.go
index d6608465..68f694cc 100644
--- a/internal/server/server_test.go
+++ b/internal/server/server_test.go
@@ -1,4 +1,4 @@
-// Copyright 2023 EMQ Technologies Co., Ltd.
+// Copyright 2023-2024 EMQ Technologies Co., Ltd.
 //
 // Licensed under the Apache License, Version 2.0 (the "License");
 // you may not use this file except in compliance with the License.
@@ -17,13 +17,13 @@ package server
 import (
 	"context"
 	"fmt"
+	"strconv"
 	"testing"
 	"time"
 
 	"github.com/stretchr/testify/require"
 
 	"github.com/lf-edge/ekuiper/v2/internal/pkg/def"
-	"github.com/lf-edge/ekuiper/v2/internal/topo/rule"
 	"github.com/lf-edge/ekuiper/v2/pkg/cast"
 )
 
@@ -37,52 +37,49 @@ func TestHandleScheduleRule(t *testing.T) {
 	require.NoError(t, err)
 	now = now.In(cast.GetConfiguredTimeZone())
 	testcases := []struct {
-		state  string
 		begin  string
 		end    string
 		action scheduleRuleAction
 	}{
 		{
-			state:  "Running",
 			begin:  "2006-01-02 15:04:01",
 			end:    "2006-01-02 15:04:06",
-			action: scheduleRuleActionDoNothing,
+			action: scheduleRuleActionStart,
 		},
 		{
-			state:  rule.RuleWait,
 			begin:  "2006-01-02 15:04:01",
 			end:    "2006-01-02 15:04:06",
 			action: scheduleRuleActionStart,
 		},
 		{
-			state:  rule.RuleTerminated,
 			begin:  "2006-01-02 15:04:01",
 			end:    "2006-01-02 15:04:04",
-			action: scheduleRuleActionDoNothing,
+			action: scheduleRuleActionStop,
 		},
 		{
-			state:  rule.RuleStarted,
 			begin:  "2006-01-02 15:04:01",
 			end:    "2006-01-02 15:04:04",
 			action: scheduleRuleActionStop,
 		},
 	}
 	for i, tc := range testcases {
-		r := &def.Rule{
-			Triggered: true,
-			Options: &def.RuleOption{
-				Cron:     "",
-				Duration: "",
-				CronDatetimeRange: []def.DatetimeRange{
-					{
-						Begin: tc.begin,
-						End:   tc.end,
+		t.Run(strconv.Itoa(i), func(t *testing.T) {
+			r := &def.Rule{
+				Triggered: true,
+				Options: &def.RuleOption{
+					Cron:     "",
+					Duration: "",
+					CronDatetimeRange: []def.DatetimeRange{
+						{
+							Begin: tc.begin,
+							End:   tc.end,
+						},
 					},
 				},
-			},
-		}
-		scheduleRuleSignal := handleScheduleRule(now, r, tc.state)
-		require.Equal(t, tc.action, scheduleRuleSignal, fmt.Sprintf("case %v", i))
+			}
+			scheduleRuleSignal := handleScheduleRule(now, r)
+			require.Equal(t, tc.action, scheduleRuleSignal, fmt.Sprintf("case %v", i))
+		})
 	}
 }
 
@@ -103,24 +100,24 @@ func TestHandleScheduleRuleState(t *testing.T) {
 	r.Options = &def.RuleOption{}
 	now, err := time.Parse("2006-01-02 15:04:05", "2006-01-02 15:04:05")
 	require.NoError(t, err)
-	require.NoError(t, handleScheduleRuleState(now, r, rule.RuleStarted))
-	require.NoError(t, handleScheduleRuleState(now, r, rule.RuleWait))
+	require.NoError(t, handleScheduleRuleState(now, r))
+	require.NoError(t, handleScheduleRuleState(now, r))
 	r.Options.CronDatetimeRange = []def.DatetimeRange{
 		{
 			Begin: "2006-01-02 15:04:01",
 			End:   "2006-01-02 15:04:06",
 		},
 	}
-	require.NoError(t, handleScheduleRuleState(now, r, rule.RuleStarted))
-	require.NoError(t, handleScheduleRuleState(now, r, rule.RuleWait))
+	require.NoError(t, handleScheduleRuleState(now, r))
+	require.NoError(t, handleScheduleRuleState(now, r))
 	r.Options.CronDatetimeRange = []def.DatetimeRange{
 		{
 			Begin: "2006-01-02 15:04:01",
 			End:   "2006-01-02 15:04:02",
 		},
 	}
-	require.NoError(t, handleScheduleRuleState(now, r, rule.RuleStarted))
-	require.NoError(t, handleScheduleRuleState(now, r, rule.RuleWait))
+	require.NoError(t, handleScheduleRuleState(now, r))
+	require.NoError(t, handleScheduleRuleState(now, r))
 }
 
 func TestStartCPUProfiling(t *testing.T) {
diff --git a/internal/server/yaml_import_export_test.go b/internal/server/yaml_import_export_test.go
index b0681e78..4dec4c62 100644
--- a/internal/server/yaml_import_export_test.go
+++ b/internal/server/yaml_import_export_test.go
@@ -55,7 +55,7 @@ func TestMetaConfiguration(t *testing.T) {
 }
 
 func prepare(t *testing.T) {
-	deleteRule("metaConf")
+	_ = registry.DeleteRule("metaConf")
 	ruleProcessor.ExecDrop("metaConf")
 	streamProcessor.ExecStreamSql(`drop stream metaConfTest`)
 	require.NoError(t, conf.WriteCfgIntoKVStorage("sources", "mqtt", "demo1", map[string]any{
@@ -70,7 +70,7 @@ func prepare(t *testing.T) {
 	_, err := streamProcessor.ExecStreamSql(`create stream metaConfTest() WITH (DATASOURCE="/API/DATA",CONF_KEY="demo1")`)
 	require.NoError(t, err)
 	rulejson := `{"trigger":false,"id":"metaConf","sql":"select * from metaConfTest","actions":[{"log":{}}]}`
-	_, err = createRule("metaConf", rulejson)
+	_, err = registry.CreateRule("metaConf", rulejson)
 	require.NoError(t, err)
 }
 
diff --git a/internal/topo/rule/ruleState_test.go b/internal/topo/rule/ruleState_test.go
index 717d0fec..f4e4cdec 100644
--- a/internal/topo/rule/ruleState_test.go
+++ b/internal/topo/rule/ruleState_test.go
@@ -1,4 +1,4 @@
-// Copyright 2022-2023 EMQ Technologies Co., Ltd.
+// Copyright 2022-2024 EMQ Technologies Co., Ltd.
 //
 // Licensed under the Apache License, Version 2.0 (the "License");
 // you may not use this file except in compliance with the License.
@@ -15,588 +15,248 @@
 package rule
 
 import (
-	"errors"
-	"fmt"
 	"sync"
 	"testing"
 	"time"
 
 	"github.com/stretchr/testify/assert"
-	"github.com/stretchr/testify/require"
 
-	"github.com/lf-edge/ekuiper/v2/internal/conf"
 	"github.com/lf-edge/ekuiper/v2/internal/pkg/def"
 	"github.com/lf-edge/ekuiper/v2/internal/processor"
 	"github.com/lf-edge/ekuiper/v2/internal/testx"
-	"github.com/lf-edge/ekuiper/v2/pkg/cast"
-	"github.com/lf-edge/ekuiper/v2/pkg/errorx"
+	"github.com/lf-edge/ekuiper/v2/pkg/timex"
 )
 
-var defaultOption = &def.RuleOption{
-	IsEventTime:        false,
-	LateTol:            cast.DurationConf(time.Second),
-	Concurrency:        1,
-	BufferLength:       1024,
-	SendMetaToSink:     false,
-	SendError:          true,
-	Qos:                def.AtMostOnce,
-	CheckpointInterval: cast.DurationConf(5 * time.Minute),
-	RestartStrategy: &def.RestartStrategy{
-		Attempts:     0,
-		Delay:        cast.DurationConf(time.Second),
-		Multiplier:   2,
-		MaxDelay:     cast.DurationConf(30 * time.Second),
-		JitterFactor: 0.1,
-	},
-}
-
 func init() {
-	testx.InitEnv("rule")
+	testx.InitEnv("Rule")
 }
 
-func TestCreate(t *testing.T) {
+func TestAPIs(t *testing.T) {
 	sp := processor.NewStreamProcessor()
-	_, err := sp.ExecStmt(`CREATE STREAM demo () WITH (DATASOURCE="users", FORMAT="JSON")`)
+	_, err := sp.ExecStmt(`CREATE STREAM demo () WITH (FORMAT="JSON", TYPE="memory", DATASOURCE="test")`)
 	assert.NoError(t, err)
 	defer sp.ExecStmt(`DROP STREAM demo`)
-	tests := []struct {
-		r    *def.Rule
-		e    error
-		code errorx.ErrorCode
-	}{
-		{
-			r: &def.Rule{
-				Triggered: false,
-				Id:        "test",
-				Sql:       "SELECT ts FROM demo",
-				Actions: []map[string]interface{}{
-					{
-						"log": map[string]interface{}{},
-					},
-				},
-				Options: defaultOption,
+	r := def.GetDefaultRule("testAPI", "select * from demo")
+	// Init state
+	st := NewState(r)
+	assert.Equal(t, r, st.Rule)
+	assert.NotNil(t, st.logger)
+	assert.Equal(t, Stopped, st.currentState)
+	assert.Equal(t, 0, len(st.actionQ))
+	assert.Equal(t, "", st.GetLastWill())
+	err = st.ResetStreamOffset("test", nil)
+	assert.EqualError(t, err, "topo is not initialized, check rule status")
+	assert.Nil(t, st.GetStreams())
+	k, v := st.GetMetrics()
+	assert.Nil(t, k)
+	assert.Nil(t, v)
+	topo := st.GetTopoGraph()
+	assert.Nil(t, topo)
+	sm := st.GetStatusMessage()
+	assert.Equal(t, "{\n  \"status\": \"Stopped\",\n  \"message\": \"\",\n  \"lastStartTimestamp\": \"0\",\n  \"lastStopTimestamp\": \"0\",\n  \"nextStartTimestamp\": \"0\"\n}", sm)
+	// Start the rule
+	e := st.Start()
+	assert.NoError(t, e)
+	time.Sleep(100 * time.Millisecond)
+	assert.Equal(t, Running, st.GetState())
+	err = st.ResetStreamOffset("test", nil)
+	assert.EqualError(t, err, "stream test not found in topo")
+	assert.Equal(t, []string{"demo"}, st.GetStreams())
+	topo = st.GetTopoGraph()
+	expTopo := &def.PrintableTopo{
+		Sources: []string{"source_demo"},
+		Edges: map[string][]any{
+			"source_demo": {
+				"op_2_project",
 			},
-			e: nil,
-		},
-		{
-			r: &def.Rule{
-				Triggered: false,
-				Id:        "test",
-				Sql:       "SELECT FROM demo",
-				Actions: []map[string]interface{}{
-					{
-						"log": map[string]interface{}{},
-					},
-				},
-				Options: defaultOption,
+			"op_2_project": {
+				"op_logToMemory_0_0_transform",
 			},
-			e:    errors.New("Parse SQL SELECT FROM demo error: found \"FROM\", expected expression.."),
-			code: errorx.ParserError,
-		},
-		{
-			r: &def.Rule{
-				Triggered: false,
-				Id:        "test",
-				Sql:       "SELECT * FROM demo1",
-				Actions: []map[string]interface{}{
-					{
-						"log": map[string]interface{}{},
-					},
-				},
-				Options: defaultOption,
+			"op_logToMemory_0_0_transform": {
+				"op_logToMemory_0_1_encode",
+			},
+			"op_logToMemory_0_1_encode": {
+				"sink_logToMemory_0",
 			},
-			e:    errors.New("fail to get stream demo1, please check if stream is created"),
-			code: errorx.PlanError,
 		},
 	}
-	for _, tt := range tests {
-		t.Run(tt.r.Id, func(t *testing.T) {
-			_, err := NewRuleState(tt.r)
-			if err != nil {
-				code, ok := errorx.GetErrorCode(err)
-				if tt.code != 0 {
-					require.True(t, ok)
-					require.Equal(t, tt.code, code)
-					assert.EqualError(t, err, tt.e.Error())
-					return
-				}
-			}
-			assert.Equal(t, tt.e, err)
-		})
+	assert.Equal(t, expTopo, topo)
+	sm = st.GetStatusMessage()
+	em := "{\n  \"status\": \"Running\",\n  \"message\": \"\",\n  \"lastStartTimestamp\": \"0\",\n  \"lastStopTimestamp\": \"0\",\n  \"nextStartTimestamp\": \"0\",\n  \"source_demo_0_records_in_total\": \"0\",\n  \"source_demo_0_records_out_total\": \"0\",\n  \"source_demo_0_messages_processed_total\": \"0\",\n  \"source_demo_0_process_latency_us\": \"0\",\n  \"source_demo_0_buffer_length\": \"0\",\n  \"source_demo_0_last_invocation\": \"0\",\n  \"source_demo_0_exceptions_total\": \"0\",\n  \"source_demo_0_last_exception\": \"\",\n  \"source_demo_0_last_exception_time\": \"0\",\n  \"op_2_project_0_records_in_total\": \"0\",\n  \"op_2_project_0_records_out_total\": \"0\",\n  \"op_2_project_0_messages_processed_total\": \"0\",\n  \"op_2_project_0_process_latency_us\": \"0\",\n  \"op_2_project_0_buffer_length\": \"0\",\n  \"op_2_project_0_last_invocation\": \"0\",\n  \"op_2_project_0_exceptions_total\": \"0\",\n  \"op_2_project_0_last_exception\": \"\",\n  \"op_2_project_0_last_exception_time\": \"0\",\n  \"op_logToMemory_0_0_transform_0_records_in_total\": \"0\",\n  \"op_logToMemory_0_0_transform_0_records_out_total\": \"0\",\n  \"op_logToMemory_0_0_transform_0_messages_processed_total\": \"0\",\n  \"op_logToMemory_0_0_transform_0_process_latency_us\": \"0\",\n  \"op_logToMemory_0_0_transform_0_buffer_length\": \"0\",\n  \"op_logToMemory_0_0_transform_0_last_invocation\": \"0\",\n  \"op_logToMemory_0_0_transform_0_exceptions_total\": \"0\",\n  \"op_logToMemory_0_0_transform_0_last_exception\": \"\",\n  \"op_logToMemory_0_0_transform_0_last_exception_time\": \"0\",\n  \"op_logToMemory_0_1_encode_0_records_in_total\": \"0\",\n  \"op_logToMemory_0_1_encode_0_records_out_total\": \"0\",\n  \"op_logToMemory_0_1_encode_0_messages_processed_total\": \"0\",\n  \"op_logToMemory_0_1_encode_0_process_latency_us\": \"0\",\n  \"op_logToMemory_0_1_encode_0_buffer_length\": \"0\",\n  \"op_logToMemory_0_1_encode_0_last_invocation\": \"0\",\n  \"op_logToMemory_0_1_encode_0_exceptions_total\": \"0\",\n  \"op_logToMemory_0_1_encode_0_last_exception\": \"\",\n  \"op_logToMemory_0_1_encode_0_last_exception_time\": \"0\",\n  \"sink_logToMemory_0_0_records_in_total\": \"0\",\n  \"sink_logToMemory_0_0_records_out_total\": \"0\",\n  \"sink_logToMemory_0_0_messages_processed_total\": \"0\",\n  \"sink_logToMemory_0_0_process_latency_us\": \"0\",\n  \"sink_logToMemory_0_0_buffer_length\": \"0\",\n  \"sink_logToMemory_0_0_last_invocation\": \"0\",\n  \"sink_logToMemory_0_0_exceptions_total\": \"0\",\n  \"sink_logToMemory_0_0_last_exception\": \"\",\n  \"sink_logToMemory_0_0_last_exception_time\": \"0\"\n}"
+	assert.Equal(t, em, sm)
+	mm := st.GetStatusMap()
+	assert.True(t, len(mm) > 0)
+	// Check stop metrics
+	st.Stop()
+	assert.Equal(t, expTopo, st.GetTopoGraph())
+	assert.Equal(t, map[string]any{"lastStartTimestamp": int64(0), "lastStopTimestamp": int64(0), "message": "canceled manually", "nextStartTimestamp": int64(0), "op_2_project_0_buffer_length": int64(0), "op_2_project_0_exceptions_total": int64(0), "op_2_project_0_last_exception": "", "op_2_project_0_last_exception_time": int64(0), "op_2_project_0_last_invocation": int64(0), "op_2_project_0_messages_processed_total": int64(0), "op_2_project_0_process_latency_us": int64(0), "op_2_project_0_records_in_total": int64(0), "op_2_project_0_records_out_total": int64(0), "op_logToMemory_0_0_transform_0_buffer_length": int64(0), "op_logToMemory_0_0_transform_0_exceptions_total": int64(0), "op_logToMemory_0_0_transform_0_last_exception": "", "op_logToMemory_0_0_transform_0_last_exception_time": int64(0), "op_logToMemory_0_0_transform_0_last_invocation": int64(0), "op_logToMemory_0_0_transform_0_messages_processed_total": int64(0), "op_logToMemory_0_0_transform_0_process_latency_us": int64(0), "op_logToMemory_0_0_transform_0_records_in_total": int64(0), "op_logToMemory_0_0_transform_0_records_out_total": int64(0), "op_logToMemory_0_1_encode_0_buffer_length": int64(0), "op_logToMemory_0_1_encode_0_exceptions_total": int64(0), "op_logToMemory_0_1_encode_0_last_exception": "", "op_logToMemory_0_1_encode_0_last_exception_time": int64(0), "op_logToMemory_0_1_encode_0_last_invocation": int64(0), "op_logToMemory_0_1_encode_0_messages_processed_total": int64(0), "op_logToMemory_0_1_encode_0_process_latency_us": int64(0), "op_logToMemory_0_1_encode_0_records_in_total": int64(0), "op_logToMemory_0_1_encode_0_records_out_total": int64(0), "sink_logToMemory_0_0_buffer_length": int64(0), "sink_logToMemory_0_0_exceptions_total": int64(0), "sink_logToMemory_0_0_last_exception": "", "sink_logToMemory_0_0_last_exception_time": int64(0), "sink_logToMemory_0_0_last_invocation": int64(0), "sink_logToMemory_0_0_messages_processed_total": int64(0), "sink_logToMemory_0_0_process_latency_us": int64(0), "sink_logToMemory_0_0_records_in_total": int64(0), "sink_logToMemory_0_0_records_out_total": int64(0), "source_demo_0_buffer_length": int64(0), "source_demo_0_exceptions_total": int64(0), "source_demo_0_last_exception": "", "source_demo_0_last_exception_time": int64(0), "source_demo_0_last_invocation": int64(0), "source_demo_0_messages_processed_total": int64(0), "source_demo_0_process_latency_us": int64(0), "source_demo_0_records_in_total": int64(0), "source_demo_0_records_out_total": int64(0), "status": "Stopped"}, st.GetStatusMap())
+	em = "{\n  \"status\": \"Stopped\",\n  \"message\": \"canceled manually\",\n  \"lastStartTimestamp\": \"0\",\n  \"lastStopTimestamp\": \"0\",\n  \"nextStartTimestamp\": \"0\",\n  \"source_demo_0_records_in_total\": \"0\",\n  \"source_demo_0_records_out_total\": \"0\",\n  \"source_demo_0_messages_processed_total\": \"0\",\n  \"source_demo_0_process_latency_us\": \"0\",\n  \"source_demo_0_buffer_length\": \"0\",\n  \"source_demo_0_last_invocation\": \"0\",\n  \"source_demo_0_exceptions_total\": \"0\",\n  \"source_demo_0_last_exception\": \"\",\n  \"source_demo_0_last_exception_time\": \"0\",\n  \"op_2_project_0_records_in_total\": \"0\",\n  \"op_2_project_0_records_out_total\": \"0\",\n  \"op_2_project_0_messages_processed_total\": \"0\",\n  \"op_2_project_0_process_latency_us\": \"0\",\n  \"op_2_project_0_buffer_length\": \"0\",\n  \"op_2_project_0_last_invocation\": \"0\",\n  \"op_2_project_0_exceptions_total\": \"0\",\n  \"op_2_project_0_last_exception\": \"\",\n  \"op_2_project_0_last_exception_time\": \"0\",\n  \"op_logToMemory_0_0_transform_0_records_in_total\": \"0\",\n  \"op_logToMemory_0_0_transform_0_records_out_total\": \"0\",\n  \"op_logToMemory_0_0_transform_0_messages_processed_total\": \"0\",\n  \"op_logToMemory_0_0_transform_0_process_latency_us\": \"0\",\n  \"op_logToMemory_0_0_transform_0_buffer_length\": \"0\",\n  \"op_logToMemory_0_0_transform_0_last_invocation\": \"0\",\n  \"op_logToMemory_0_0_transform_0_exceptions_total\": \"0\",\n  \"op_logToMemory_0_0_transform_0_last_exception\": \"\",\n  \"op_logToMemory_0_0_transform_0_last_exception_time\": \"0\",\n  \"op_logToMemory_0_1_encode_0_records_in_total\": \"0\",\n  \"op_logToMemory_0_1_encode_0_records_out_total\": \"0\",\n  \"op_logToMemory_0_1_encode_0_messages_processed_total\": \"0\",\n  \"op_logToMemory_0_1_encode_0_process_latency_us\": \"0\",\n  \"op_logToMemory_0_1_encode_0_buffer_length\": \"0\",\n  \"op_logToMemory_0_1_encode_0_last_invocation\": \"0\",\n  \"op_logToMemory_0_1_encode_0_exceptions_total\": \"0\",\n  \"op_logToMemory_0_1_encode_0_last_exception\": \"\",\n  \"op_logToMemory_0_1_encode_0_last_exception_time\": \"0\",\n  \"sink_logToMemory_0_0_records_in_total\": \"0\",\n  \"sink_logToMemory_0_0_records_out_total\": \"0\",\n  \"sink_logToMemory_0_0_messages_processed_total\": \"0\",\n  \"sink_logToMemory_0_0_process_latency_us\": \"0\",\n  \"sink_logToMemory_0_0_buffer_length\": \"0\",\n  \"sink_logToMemory_0_0_last_invocation\": \"0\",\n  \"sink_logToMemory_0_0_exceptions_total\": \"0\",\n  \"sink_logToMemory_0_0_last_exception\": \"\",\n  \"sink_logToMemory_0_0_last_exception_time\": \"0\"\n}"
+	assert.Equal(t, em, st.GetStatusMessage())
+	assert.Equal(t, Stopped, st.currentState)
+	// Update rule
+	st.Rule = def.GetDefaultRule("testAPI", "select abc from demo where a > 3")
+	e = st.Validate()
+	assert.NoError(t, e)
+	e = st.Start()
+	assert.NoError(t, e)
+	time.Sleep(100 * time.Millisecond)
+	assert.Equal(t, Running, st.currentState)
+	err = st.ResetStreamOffset("test", nil)
+	assert.EqualError(t, err, "stream test not found in topo")
+	assert.Equal(t, []string{"demo"}, st.GetStreams())
+	topo = st.GetTopoGraph()
+	expTopo = &def.PrintableTopo{
+		Sources: []string{"source_demo"},
+		Edges: map[string][]any{
+			"source_demo": {
+				"op_2_filter",
+			},
+			"op_2_filter": {
+				"op_3_project",
+			},
+			"op_3_project": {
+				"op_logToMemory_0_0_transform",
+			},
+			"op_logToMemory_0_0_transform": {
+				"op_logToMemory_0_1_encode",
+			},
+			"op_logToMemory_0_1_encode": {
+				"sink_logToMemory_0",
+			},
+		},
 	}
+	assert.Equal(t, expTopo, topo)
+	sm = st.GetStatusMessage()
+	em = "{\n  \"status\": \"Running\",\n  \"message\": \"\",\n  \"lastStartTimestamp\": \"0\",\n  \"lastStopTimestamp\": \"0\",\n  \"nextStartTimestamp\": \"0\",\n  \"source_demo_0_records_in_total\": \"0\",\n  \"source_demo_0_records_out_total\": \"0\",\n  \"source_demo_0_messages_processed_total\": \"0\",\n  \"source_demo_0_process_latency_us\": \"0\",\n  \"source_demo_0_buffer_length\": \"0\",\n  \"source_demo_0_last_invocation\": \"0\",\n  \"source_demo_0_exceptions_total\": \"0\",\n  \"source_demo_0_last_exception\": \"\",\n  \"source_demo_0_last_exception_time\": \"0\",\n  \"op_2_filter_0_records_in_total\": \"0\",\n  \"op_2_filter_0_records_out_total\": \"0\",\n  \"op_2_filter_0_messages_processed_total\": \"0\",\n  \"op_2_filter_0_process_latency_us\": \"0\",\n  \"op_2_filter_0_buffer_length\": \"0\",\n  \"op_2_filter_0_last_invocation\": \"0\",\n  \"op_2_filter_0_exceptions_total\": \"0\",\n  \"op_2_filter_0_last_exception\": \"\",\n  \"op_2_filter_0_last_exception_time\": \"0\",\n  \"op_3_project_0_records_in_total\": \"0\",\n  \"op_3_project_0_records_out_total\": \"0\",\n  \"op_3_project_0_messages_processed_total\": \"0\",\n  \"op_3_project_0_process_latency_us\": \"0\",\n  \"op_3_project_0_buffer_length\": \"0\",\n  \"op_3_project_0_last_invocation\": \"0\",\n  \"op_3_project_0_exceptions_total\": \"0\",\n  \"op_3_project_0_last_exception\": \"\",\n  \"op_3_project_0_last_exception_time\": \"0\",\n  \"op_logToMemory_0_0_transform_0_records_in_total\": \"0\",\n  \"op_logToMemory_0_0_transform_0_records_out_total\": \"0\",\n  \"op_logToMemory_0_0_transform_0_messages_processed_total\": \"0\",\n  \"op_logToMemory_0_0_transform_0_process_latency_us\": \"0\",\n  \"op_logToMemory_0_0_transform_0_buffer_length\": \"0\",\n  \"op_logToMemory_0_0_transform_0_last_invocation\": \"0\",\n  \"op_logToMemory_0_0_transform_0_exceptions_total\": \"0\",\n  \"op_logToMemory_0_0_transform_0_last_exception\": \"\",\n  \"op_logToMemory_0_0_transform_0_last_exception_time\": \"0\",\n  \"op_logToMemory_0_1_encode_0_records_in_total\": \"0\",\n  \"op_logToMemory_0_1_encode_0_records_out_total\": \"0\",\n  \"op_logToMemory_0_1_encode_0_messages_processed_total\": \"0\",\n  \"op_logToMemory_0_1_encode_0_process_latency_us\": \"0\",\n  \"op_logToMemory_0_1_encode_0_buffer_length\": \"0\",\n  \"op_logToMemory_0_1_encode_0_last_invocation\": \"0\",\n  \"op_logToMemory_0_1_encode_0_exceptions_total\": \"0\",\n  \"op_logToMemory_0_1_encode_0_last_exception\": \"\",\n  \"op_logToMemory_0_1_encode_0_last_exception_time\": \"0\",\n  \"sink_logToMemory_0_0_records_in_total\": \"0\",\n  \"sink_logToMemory_0_0_records_out_total\": \"0\",\n  \"sink_logToMemory_0_0_messages_processed_total\": \"0\",\n  \"sink_logToMemory_0_0_process_latency_us\": \"0\",\n  \"sink_logToMemory_0_0_buffer_length\": \"0\",\n  \"sink_logToMemory_0_0_last_invocation\": \"0\",\n  \"sink_logToMemory_0_0_exceptions_total\": \"0\",\n  \"sink_logToMemory_0_0_last_exception\": \"\",\n  \"sink_logToMemory_0_0_last_exception_time\": \"0\"\n}"
+	assert.Equal(t, em, sm)
+	e = st.Delete()
+	assert.NoError(t, e)
 }
 
-func TestUpdate(t *testing.T) {
-	ignoreSignal = true
-	defer func() {
-		ignoreSignal = false
-	}()
+func TestStateTransit(t *testing.T) {
 	sp := processor.NewStreamProcessor()
-	sp.ExecStmt(`CREATE STREAM demo () WITH (DATASOURCE="users", FORMAT="JSON")`)
+	_, err := sp.ExecStmt(`CREATE STREAM demo () WITH (FORMAT="JSON", TYPE="memory", DATASOURCE="test")`)
+	assert.NoError(t, err)
 	defer sp.ExecStmt(`DROP STREAM demo`)
 	tests := []struct {
-		r         *def.Rule
-		e         error
-		triggered int
-		code      errorx.ErrorCode
+		name       string
+		r          *def.Rule
+		actions    []ActionSignal
+		async      bool
+		finalState RunState
 	}{
 		{
-			r: &def.Rule{
-				Triggered: false,
-				Id:        "test",
-				Sql:       "SELECT FROM demo",
-				Actions: []map[string]interface{}{
-					{
-						"log": map[string]interface{}{},
-					},
-				},
-				Options: defaultOption,
-			},
-			e:         errors.New("Parse SQL SELECT FROM demo error: found \"FROM\", expected expression.."),
-			code:      errorx.ParserError,
-			triggered: 1,
-		},
-		{
-			r: &def.Rule{
-				Triggered: false,
-				Id:        "test",
-				Sql:       "SELECT * FROM demo1",
-				Actions: []map[string]interface{}{
-					{
-						"log": map[string]interface{}{},
-					},
-				},
-				Options: defaultOption,
-			},
-			e:         errors.New("fail to get stream demo1, please check if stream is created"),
-			code:      errorx.PlanError,
-			triggered: 1,
+			name:       "fast start stop",
+			r:          def.GetDefaultRule("testNormal", "select * from demo"),
+			actions:    []ActionSignal{ActionSignalStart, ActionSignalStart, ActionSignalStop, ActionSignalStart, ActionSignalStart, ActionSignalStop},
+			finalState: Stopped,
 		},
 		{
-			r: &def.Rule{
-				Triggered: true,
-				Id:        "test",
-				Sql:       "SELECT * FROM demo",
-				Actions: []map[string]interface{}{
-					{
-						"log": map[string]interface{}{},
-					},
-				},
-				Options: defaultOption,
-			},
-			e:         nil,
-			triggered: 1,
+			name:       "async fast start stop",
+			r:          def.GetDefaultRule("testAsync", "select * from demo"),
+			actions:    []ActionSignal{ActionSignalStart, ActionSignalStop, ActionSignalStop, ActionSignalStart, ActionSignalStop, ActionSignalStop, ActionSignalStart},
+			finalState: Running,
+			async:      true,
 		},
 		{
-			r: &def.Rule{
-				Triggered: false,
-				Id:        "test",
-				Sql:       "SELECT * FROM demo",
-				Actions: []map[string]interface{}{
-					{
-						"log": map[string]interface{}{},
-					},
-				},
-				Options: defaultOption,
-			},
-			e:         nil,
-			triggered: 0,
+			name:       "invalid",
+			r:          def.GetDefaultRule("testAsync", "select * from demo2"),
+			actions:    []ActionSignal{ActionSignalStart, ActionSignalStop, ActionSignalStop, ActionSignalStart, ActionSignalStop, ActionSignalStop, ActionSignalStart},
+			finalState: StoppedByErr,
+			async:      true,
 		},
 	}
-	for i, tt := range tests {
-		rs, err := NewRuleState(&def.Rule{
-			Triggered: true,
-			Id:        "test",
-			Sql:       "SELECT ts FROM demo",
-			Actions: []map[string]interface{}{
-				{
-					"log": map[string]interface{}{},
-				},
-			},
-			Options: defaultOption,
-		})
-		require.NoError(t, err)
-		err = rs.Start()
-		require.NoError(t, err)
-		time.Sleep(5 * time.Millisecond)
-		require.Equal(t, 1, rs.triggered, fmt.Sprintf("case %v failed", i))
-		err = rs.UpdateTopo(tt.r)
-		time.Sleep(5 * time.Millisecond)
-		if err != nil {
-			code, ok := errorx.GetErrorCode(err)
-			if tt.code != 0 {
-				require.True(t, ok)
-				require.Equal(t, tt.code, code)
-				require.Equal(t, tt.e.Error(), err.Error(), fmt.Sprintf("case %v failed", i))
-				continue
+	for _, v := range tests {
+		t.Run(v.name, func(t *testing.T) {
+			st := NewState(v.r)
+			defer st.Delete()
+			st.actionQ = []ActionSignal{ActionSignalStart}
+			var wg sync.WaitGroup
+			if v.async {
+				wg.Add(len(v.actions) - 1)
 			}
-		}
-		require.Equal(t, tt.triggered, rs.triggered, fmt.Sprintf("case %v failed", i))
-		rs.Close()
-	}
-}
-
-func TestUpdateScheduleRule(t *testing.T) {
-	sp := processor.NewStreamProcessor()
-	sp.ExecStmt(`CREATE STREAM demo () WITH (DATASOURCE="users", FORMAT="JSON")`)
-	defer sp.ExecStmt(`DROP STREAM demo`)
-	scheduleOption1 := *defaultOption
-	scheduleOption1.Cron = "mockCron"
-	scheduleOption1.Duration = "1s"
-	rule1 := &def.Rule{
-		Triggered: true,
-		Id:        "test",
-		Sql:       "SELECT ts FROM demo",
-		Actions: []map[string]interface{}{
-			{
-				"log": map[string]interface{}{},
-			},
-		},
-		Options: &scheduleOption1,
-	}
-	rs, err := NewRuleState(rule1)
-	require.NoError(t, err)
-	defer rs.Close()
-	err = rs.startScheduleRule()
-	require.NoError(t, err)
-	require.True(t, rs.cronState.isInSchedule)
-	require.Equal(t, "mockCron", rs.cronState.cron)
-	require.Equal(t, "1s", rs.cronState.duration)
-
-	scheduleOption2 := *defaultOption
-	scheduleOption2.Cron = "mockCron2"
-	scheduleOption2.Duration = "2s"
-	rule2 := &def.Rule{
-		Triggered: true,
-		Id:        "test",
-		Sql:       "SELECT ts FROM demo",
-		Actions: []map[string]interface{}{
-			{
-				"log": map[string]interface{}{},
-			},
-		},
-		Options: &scheduleOption2,
+			for i, a := range v.actions {
+				if i == len(v.actions)-1 {
+					break
+				}
+				if v.async {
+					go func() {
+						sendAction(st, a)
+						wg.Done()
+					}()
+				} else {
+					sendAction(st, a)
+				}
+			}
+			if v.async {
+				wg.Wait()
+			}
+			sendAction(st, v.actions[len(v.actions)-1])
+			time.Sleep(100 * time.Millisecond)
+			assert.Equal(t, v.finalState, st.GetState())
+		})
 	}
-	err = rs.UpdateTopo(rule2)
-	require.NoError(t, err)
-	require.Equal(t, "mockCron2", rs.cronState.cron)
-	require.Equal(t, "2s", rs.cronState.duration)
 }
 
-func TestMultipleAccess(t *testing.T) {
+func TestLongScheduleTransit(t *testing.T) {
 	sp := processor.NewStreamProcessor()
-	sp.ExecStmt(`CREATE STREAM demo () WITH (DATASOURCE="users", FORMAT="JSON")`)
-	defer sp.ExecStmt(`DROP STREAM demo`)
-	rs, err := NewRuleState(&def.Rule{
-		Triggered: false,
-		Id:        "test",
-		Sql:       "SELECT ts FROM demo",
-		Actions: []map[string]interface{}{
-			{
-				"log": map[string]interface{}{},
-			},
-		},
-		Options: defaultOption,
-	})
-	if err != nil {
-		t.Error(err)
-		return
-	}
-	defer rs.Close()
-	err = rs.Start()
-	if err != nil {
-		t.Error(err)
-		return
-	}
-	var wg sync.WaitGroup
-	wg.Add(10)
-	for i := 0; i < 10; i++ {
-		if i%3 == 0 {
-			go func(i int) {
-				rs.Stop()
-				fmt.Printf("%d:%d\n", i, rs.triggered)
-				wg.Done()
-			}(i)
-		} else {
-			go func(i int) {
-				rs.Start()
-				fmt.Printf("%d:%d\n", i, rs.triggered)
-				wg.Done()
-			}(i)
-		}
-	}
-	wg.Wait()
-	fmt.Printf("before %d:%d\n", 10, rs.triggered)
-	err = rs.Stop()
+	_, err := sp.ExecStmt(`CREATE STREAM demo () WITH (FORMAT="JSON", TYPE="memory", DATASOURCE="test")`)
 	assert.NoError(t, err)
-	fmt.Printf("%d:%d\n", 10, rs.triggered)
-	if rs.triggered != 0 {
-		t.Errorf("triggered mismatch:\n\nexp=%#v\n\ngot=%#v\n\n", 0, rs.triggered)
-	}
-}
-
-// Test rule state message
-func TestRuleState_Start(t *testing.T) {
-	sp := processor.NewStreamProcessor()
-	sp.ExecStmt(`CREATE STREAM demo () WITH (TYPE="memory", FORMAT="JSON", DATASOURCE="test")`)
 	defer sp.ExecStmt(`DROP STREAM demo`)
-	// Test rule not triggered
-	r := &def.Rule{
-		Triggered: false,
-		Id:        "test",
-		Sql:       "SELECT ts FROM demo",
-		Actions: []map[string]interface{}{
-			{
-				"log": map[string]interface{}{},
-			},
+	// set now for schedule rule
+	now := time.Date(2024, time.August, 8, 15, 38, 0, 0, time.UTC)
+	timex.Set(now.UnixMilli())
+	sr := def.GetDefaultRule("testScheduleNotIn", "select * from demo")
+	sr.Options.CronDatetimeRange = []def.DatetimeRange{
+		{
+			Begin: "2024-08-08 16:04:01",
+			End:   "2024-08-08 16:30:01",
 		},
-		Options: defaultOption,
 	}
-	const ruleStopped = "Stopped: canceled manually."
-	const ruleStarted = "Running"
-	// TODO enable this again
-	//t.Run("test rule loaded but not started", func(t *testing.T) {
-	//	rs, err := NewRuleState(r)
-	//	if err != nil {
-	//		t.Error(err)
-	//		return
-	//	}
-	//	state, err := rs.GetState()
-	//	if err != nil {
-	//		t.Errorf("get rule state error: %v", err)
-	//		return
-	//	}
-	//	if state != ruleStopped {
-	//		t.Errorf("rule state mismatch: exp=%v, got=%v", ruleStopped, state)
-	//		return
-	//	}
-	//})
-	t.Run("test rule started", func(t *testing.T) {
-		rs, err := NewRuleState(r)
-		if err != nil {
-			t.Error(err)
-			return
-		}
-		err = rs.Start()
-		if err != nil {
-			t.Error(err)
-			return
-		}
-		time.Sleep(100 * time.Millisecond)
-		state, err := rs.GetState()
-		if err != nil {
-			t.Errorf("get rule state error: %v", err)
-			return
-		}
-		if state != ruleStarted {
-			t.Errorf("rule state mismatch: exp=%v, got=%v", ruleStopped, state)
-			return
-		}
-	})
-	t.Run("test rule loaded and stopped", func(t *testing.T) {
-		rs, err := NewRuleState(r)
-		if err != nil {
-			t.Error(err)
-			return
-		}
-		err = rs.Start()
-		if err != nil {
-			t.Error(err)
-			return
-		}
-		err = rs.Close()
-		if err != nil {
-			t.Error(err)
-			return
-		}
-		state, err := rs.GetState()
-		if err != nil {
-			t.Errorf("get rule state error: %v", err)
-			return
-		}
-		if state != ruleStopped {
-			t.Errorf("rule state mismatch: exp=%v, got=%v", ruleStopped, state)
-			return
-		}
-	})
+	st := NewState(sr)
+	defer st.Delete()
+	// Start run, but not in schedule
+	e := st.Start()
+	assert.NoError(t, e)
+	assert.Equal(t, ScheduledStop, st.GetState())
+	// Scheduled stop to start, no change
+	_ = st.Start()
+	assert.Equal(t, ScheduledStop, st.GetState())
+	// Scheduled stop to stop, stop
+	st.Stop()
+	assert.Equal(t, Stopped, st.GetState())
+	// Time move to schedule, should start
+	timex.Add(30 * time.Minute)
+	_ = st.ScheduleStart()
+	// Notice: mock the action queue. The action must be the same as the next otherwise it will loop infinitely
+	st.actionQ = append(st.actionQ, ActionSignalScheduledStart)
+	_ = st.ScheduleStart()
+	time.Sleep(100 * time.Millisecond)
+	assert.Equal(t, Running, st.GetState())
+	// Time move out of schedule, scheduled stop
+	timex.Add(30 * time.Minute)
+	st.ScheduleStop()
+	// Notice: mock the action queue. The action must be the same as the next otherwise it will loop infinitely
+	st.actionQ = append(st.actionQ, ActionSignalScheduledStop)
+	time.Sleep(100 * time.Millisecond)
+	assert.Equal(t, ScheduledStop, st.GetState())
+	st.ScheduleStop()
+	assert.Equal(t, ScheduledStop, st.GetState())
 }
 
-// TODO test schedule rule again
-func TestScheduleRule(t *testing.T) {
-	conf.IsTesting = true
-	sp := processor.NewStreamProcessor()
-	sp.ExecStmt(`CREATE STREAM demo () WITH (TYPE="memory", DATASOURCE="test", FORMAT="JSON")`)
-	defer sp.ExecStmt(`DROP STREAM demo`)
-	// Test rule not triggered
-	r := &def.Rule{
-		Triggered: false,
-		Id:        "test",
-		Sql:       "SELECT ts FROM demo",
-		Actions: []map[string]interface{}{
-			{
-				"log": map[string]interface{}{},
-			},
-		},
-		Options: defaultOption,
+func sendAction(st *State, a ActionSignal) {
+	switch a {
+	case ActionSignalStart:
+		_ = st.Start()
+	case ActionSignalStop:
+		st.Stop()
+	case ActionSignalScheduledStart:
+		_ = st.ScheduleStart()
+	case ActionSignalScheduledStop:
+		st.ScheduleStop()
 	}
-	r.Options.Cron = "mockCron"
-	r.Options.Duration = "1s"
-	const ruleStarted = "Running"
-	const ruleStopped = "Stopped: waiting for next schedule."
-	func() {
-		rs, err := NewRuleState(r)
-		if err != nil {
-			t.Error(err)
-			return
-		}
-		if err := rs.startScheduleRule(); err != nil {
-			t.Error(err)
-			return
-		}
-		time.Sleep(500 * time.Millisecond)
-		state, err := rs.GetState()
-		if err != nil {
-			t.Errorf("get rule state error: %v", err)
-			return
-		}
-		if state != ruleStarted {
-			t.Errorf("rule state mismatch: exp=%v, got=%v", ruleStarted, state)
-			return
-		}
-		if !rs.cronState.isInSchedule {
-			t.Error("cron state should be in schedule")
-			return
-		}
-	}()
-
-	func() {
-		rs, err := NewRuleState(r)
-		if err != nil {
-			t.Error(err)
-			return
-		}
-		if err := rs.startScheduleRule(); err != nil {
-			t.Error(err)
-			return
-		}
-		time.Sleep(1500 * time.Millisecond)
-		state, err := rs.GetState()
-		if err != nil {
-			t.Errorf("get rule state error: %v", err)
-			return
-		}
-		if state != ruleStopped {
-			t.Errorf("rule state mismatch: exp=%v, got=%v", ruleStopped, state)
-			return
-		}
-		if !rs.cronState.isInSchedule {
-			t.Error("cron state should be in schedule")
-			return
-		}
-	}()
-
-	func() {
-		rs, err := NewRuleState(r)
-		if err != nil {
-			t.Error(err)
-			return
-		}
-		if err := rs.startScheduleRule(); err != nil {
-			t.Error(err)
-			return
-		}
-		if err := rs.startScheduleRule(); err == nil {
-			t.Error("rule can't be register in cron twice")
-			return
-		} else {
-			if err.Error() != "rule test is already in schedule" {
-				t.Error("error message wrong")
-				return
-			}
-		}
-	}()
-
-	func() {
-		rs, err := NewRuleState(r)
-		if err != nil {
-			t.Error(err)
-			return
-		}
-		if err := rs.startScheduleRule(); err != nil {
-			t.Error(err)
-			return
-		}
-		if err := rs.Stop(); err != nil {
-			t.Error(err)
-			return
-		}
-		state, err := rs.GetState()
-		if err != nil {
-			t.Errorf("get rule state error: %v", err)
-			return
-		}
-		if state != "Stopped: canceled manually." {
-			t.Errorf("rule state mismatch: exp=%v, got=%v", "Stopped: canceled manually.", state)
-			return
-		}
-		if rs.cronState.isInSchedule {
-			t.Error("cron state shouldn't be in schedule")
-			return
-		}
-	}()
-
-	func() {
-		rs, err := NewRuleState(r)
-		if err != nil {
-			t.Error(err)
-			return
-		}
-		if err := rs.Stop(); err != nil {
-			t.Error(err)
-			return
-		}
-		if err := rs.Close(); err != nil {
-			t.Error(err)
-			return
-		}
-	}()
 }
 
-const layout = "2006-01-02 15:04:05"
-
-func TestRuleStateInternalStop(t *testing.T) {
-	conf.IsTesting = true
-	sp := processor.NewStreamProcessor()
-	sp.ExecStmt(`CREATE STREAM demo () WITH (TYPE="memory", DATASOURCE="test", FORMAT="JSON")`)
-	defer sp.ExecStmt(`DROP STREAM demo`)
-	r := &def.Rule{
-		Triggered: false,
-		Id:        "test",
-		Sql:       "SELECT ts FROM demo",
-		Actions: []map[string]interface{}{
-			{
-				"log": map[string]interface{}{},
-			},
-		},
-		Options: defaultOption,
-	}
-	r.Options.Cron = "123"
-	rs, err := NewRuleState(r)
-	require.NoError(t, err)
-	err = rs.InternalStop()
-	require.Error(t, err)
-
-	r.Options.Cron = ""
-	r.Options.Duration = ""
-	r.Options.CronDatetimeRange = []def.DatetimeRange{
-		{
-			Begin: layout,
-			End:   layout,
-		},
-	}
-	rs, err = NewRuleState(r)
-	require.NoError(t, err)
-	err = rs.InternalStop()
-	require.NoError(t, err)
-	require.Equal(t, rs.triggered, 2)
+func TestRuleRestart(t *testing.T) {
+	// TODO added later
 }
diff --git a/test/change_rule_status.jmx b/test/change_rule_status.jmx
index 8504bc81..4b475b9d 100644
--- a/test/change_rule_status.jmx
+++ b/test/change_rule_status.jmx
@@ -1,6 +1,6 @@
 <?xml version="1.0" encoding="UTF-8"?>
 <!--
-  ~ Copyright 2021-2022 EMQ Technologies Co., Ltd.
+  ~ Copyright 2021-2024 EMQ Technologies Co., Ltd.
   ~
   ~ Licensed under the Apache License, Version 2.0 (the "License");
   ~ you may not use this file except in compliance with the License.
@@ -299,6 +299,7 @@
             <stringProp name="mqtt.message_type">String</stringProp>
             <stringProp name="mqtt.message_type_fixed_length">1024</stringProp>
             <stringProp name="mqtt.message_to_sent">{&quot;temperature&quot;: ${__Random(1,50,)}, &quot;humidity&quot; : ${__Random(1,80,)}}</stringProp>
+            <boolProp name="mqtt.retained_message">false</boolProp>
           </net.xmeter.samplers.PubSampler>
           <hashTree/>
           <HTTPSamplerProxy guiclass="HttpTestSampleGui" testclass="HTTPSamplerProxy" testname="API_GetRuleStatus" enabled="true">
@@ -409,15 +410,35 @@
             <stringProp name="HTTPSampler.response_timeout"></stringProp>
           </HTTPSamplerProxy>
           <hashTree>
-            <ResponseAssertion guiclass="AssertionGui" testclass="ResponseAssertion" testname="Response Assertion" enabled="true">
-              <collectionProp name="Asserion.test_strings">
-                <stringProp name="1607612103">Stopped: canceled manually.</stringProp>
-              </collectionProp>
-              <stringProp name="Assertion.custom_message"></stringProp>
-              <stringProp name="Assertion.test_field">Assertion.response_data</stringProp>
-              <boolProp name="Assertion.assume_success">false</boolProp>
-              <intProp name="Assertion.test_type">16</intProp>
-            </ResponseAssertion>
+            <JSONPathAssertion guiclass="JSONPathAssertionGui" testclass="JSONPathAssertion" testname="JSON Assertion"
+                               enabled="true">
+              <stringProp name="JSON_PATH">$.status</stringProp>
+              <stringProp name="EXPECTED_VALUE">Stopped</stringProp>
+              <boolProp name="JSONVALIDATION">true</boolProp>
+              <boolProp name="EXPECT_NULL">false</boolProp>
+              <boolProp name="INVERT">false</boolProp>
+              <boolProp name="ISREGEX">false</boolProp>
+            </JSONPathAssertion>
+            <hashTree/>
+            <JSONPathAssertion guiclass="JSONPathAssertionGui" testclass="JSONPathAssertion" testname="JSON Assertion"
+                               enabled="true">
+              <stringProp name="JSON_PATH">$.message</stringProp>
+              <stringProp name="EXPECTED_VALUE">canceled manually</stringProp>
+              <boolProp name="JSONVALIDATION">true</boolProp>
+              <boolProp name="EXPECT_NULL">false</boolProp>
+              <boolProp name="INVERT">false</boolProp>
+              <boolProp name="ISREGEX">false</boolProp>
+            </JSONPathAssertion>
+            <hashTree/>
+            <JSONPathAssertion guiclass="JSONPathAssertionGui" testclass="JSONPathAssertion" testname="JSON Assertion"
+                               enabled="true">
+              <stringProp name="JSON_PATH">$.sink_mqtt_0_0_records_in_total</stringProp>
+              <stringProp name="EXPECTED_VALUE">1</stringProp>
+              <boolProp name="JSONVALIDATION">true</boolProp>
+              <boolProp name="EXPECT_NULL">false</boolProp>
+              <boolProp name="INVERT">false</boolProp>
+              <boolProp name="ISREGEX">false</boolProp>
+            </JSONPathAssertion>
             <hashTree/>
           </hashTree>
         </hashTree>
@@ -589,16 +610,24 @@
             <stringProp name="HTTPSampler.response_timeout"></stringProp>
           </HTTPSamplerProxy>
           <hashTree>
-            <ResponseAssertion guiclass="AssertionGui" testclass="ResponseAssertion" testname="Response Assertion" enabled="true">
-              <collectionProp name="Asserion.test_strings">
-                <stringProp name="1607612103">Stopped: canceled manually.</stringProp>
-              </collectionProp>
-              <stringProp name="Assertion.custom_message"></stringProp>
-              <stringProp name="Assertion.test_field">Assertion.response_data</stringProp>
-              <boolProp name="Assertion.assume_success">false</boolProp>
-              <intProp name="Assertion.test_type">16</intProp>
-            </ResponseAssertion>
+            <JSONPathAssertion guiclass="JSONPathAssertionGui" testclass="JSONPathAssertion" testname="JSON Assertion" enabled="true">
+              <stringProp name="JSON_PATH">$.status</stringProp>
+              <stringProp name="EXPECTED_VALUE">Stopped</stringProp>
+              <boolProp name="JSONVALIDATION">true</boolProp>
+              <boolProp name="EXPECT_NULL">false</boolProp>
+              <boolProp name="INVERT">false</boolProp>
+              <boolProp name="ISREGEX">false</boolProp>
+            </JSONPathAssertion>
             <hashTree/>
+            <JSONPathAssertion guiclass="JSONPathAssertionGui" testclass="JSONPathAssertion" testname="JSON Assertion" enabled="true">
+              <stringProp name="JSON_PATH">$.sink_mqtt_0_0_records_in_total</stringProp>
+              <stringProp name="EXPECTED_VALUE">2</stringProp>
+              <boolProp name="JSONVALIDATION">true</boolProp>
+              <boolProp name="EXPECT_NULL">false</boolProp>
+              <boolProp name="INVERT">false</boolProp>
+              <boolProp name="ISREGEX">false</boolProp>
+            </JSONPathAssertion>
+          <hashTree/>
           </hashTree>
         </hashTree>
         <TransactionController guiclass="TransactionControllerGui" testclass="TransactionController" testname="RestartAndRun" enabled="true">
@@ -842,15 +871,25 @@
             <stringProp name="HTTPSampler.response_timeout"></stringProp>
           </HTTPSamplerProxy>
           <hashTree>
-            <ResponseAssertion guiclass="AssertionGui" testclass="ResponseAssertion" testname="Response Assertion" enabled="true">
-              <collectionProp name="Asserion.test_strings">
-                <stringProp name="1607612103">Stopped: canceled manually.</stringProp>
-              </collectionProp>
-              <stringProp name="Assertion.custom_message"></stringProp>
-              <stringProp name="Assertion.test_field">Assertion.response_data</stringProp>
-              <boolProp name="Assertion.assume_success">false</boolProp>
-              <intProp name="Assertion.test_type">16</intProp>
-            </ResponseAssertion>
+            <JSONPathAssertion guiclass="JSONPathAssertionGui" testclass="JSONPathAssertion" testname="JSON Assertion"
+                               enabled="true">
+              <stringProp name="JSON_PATH">$.status</stringProp>
+              <stringProp name="EXPECTED_VALUE">Stopped</stringProp>
+              <boolProp name="JSONVALIDATION">true</boolProp>
+              <boolProp name="EXPECT_NULL">false</boolProp>
+              <boolProp name="INVERT">false</boolProp>
+              <boolProp name="ISREGEX">false</boolProp>
+            </JSONPathAssertion>
+            <hashTree/>
+            <JSONPathAssertion guiclass="JSONPathAssertionGui" testclass="JSONPathAssertion" testname="JSON Assertion"
+                               enabled="true">
+              <stringProp name="JSON_PATH">$.sink_mqtt_0_0_records_in_total</stringProp>
+              <stringProp name="EXPECTED_VALUE">1</stringProp>
+              <boolProp name="JSONVALIDATION">true</boolProp>
+              <boolProp name="EXPECT_NULL">false</boolProp>
+              <boolProp name="INVERT">false</boolProp>
+              <boolProp name="ISREGEX">false</boolProp>
+            </JSONPathAssertion>
             <hashTree/>
           </hashTree>
         </hashTree>