Repository: lf-edge/ekuiper
Pull request #3003 (2024-07-12 09:40:11)

fix: fix restart rule state

codecov[bot]: ## [Codecov](https://app.codecov.io/gh/lf-edge/ekuiper/pull/3003?dropdown=coverage&src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=lf-edge) Report
All modified and coverable lines are covered by tests :white_check_mark:
> Project coverage is 68.54%. Comparing base [(`f2fdc02`)](https://app.codecov.io/gh/lf-edge/ekuiper/commit/f2fdc02215f9b49b9ae428e58b77376b88ef0667?dropdown=coverage&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=lf-edge) to head [(`a30d352`)](https://app.codecov.io/gh/lf-edge/ekuiper/commit/a30d352b4e73721c1abc59da9398b242f9dd820a?dropdown=coverage&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=lf-edge).
> Report is 3 commits behind head on master-v2.

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


```diff
@@              Coverage Diff              @@
##           master-v2    #3003      +/-   ##
=============================================
+ Coverage      68.28%   68.54%   +0.26%     
=============================================
  Files            344      348       +4     
  Lines          33079    33417     +338     
=============================================
+ Hits           22586    22904     +318     
+ Misses          8476     8468       -8     
- Partials        2017     2045      +28     
```

</details>

[:umbrella: View full report in Codecov by Sentry](https://app.codecov.io/gh/lf-edge/ekuiper/pull/3003?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/topo/rule/ruleState.go b/internal/topo/rule/ruleState.go
index 06d920d7..8ca05a20 100644
--- a/internal/topo/rule/ruleState.go
+++ b/internal/topo/rule/ruleState.go
@@ -128,6 +128,10 @@ func NewRuleState(rule *def.Rule) (rs *RuleState, err error) {
 		} else {
 			rs.Topology = tp
 		}
+		if !rs.Rule.Triggered {
+			// manually force stop rule
+			rs.Topology.Cancel()
+		}
 		return nil
 	})
 	rs.run()


Test patch:
diff --git a/internal/server/rest_test.go b/internal/server/rest_test.go
index 8867636c..8ba54e73 100644
--- a/internal/server/rest_test.go
+++ b/internal/server/rest_test.go
@@ -209,6 +209,45 @@ func (suite *RestTestSuite) Test_sourcesManageHandler() {
 	assert.Equal(suite.T(), http.StatusOK, w.Code)
 }
 
+func (suite *RestTestSuite) TestRecoverRule() {
+	// drop stream
+	req, _ := http.NewRequest(http.MethodDelete, "http://localhost:8080/streams/recoverTest", bytes.NewBufferString("any"))
+	w := httptest.NewRecorder()
+	suite.r.ServeHTTP(w, req)
+	req1, _ := http.NewRequest(http.MethodDelete, "http://localhost:8080/rules/recoverTest", bytes.NewBufferString("any"))
+	w1 := httptest.NewRecorder()
+	suite.r.ServeHTTP(w1, req1)
+
+	buf1 := bytes.NewBuffer([]byte(`{"sql":"CREATE stream recoverTest() WITH (DATASOURCE=\"0\", TYPE=\"mqtt\")"}`))
+	req1, _ = http.NewRequest(http.MethodPost, "http://localhost:8080/streams", buf1)
+	w1 = httptest.NewRecorder()
+	suite.r.ServeHTTP(w1, req1)
+	// create rule with trigger false
+	ruleJson := `{"id": "recoverTest","triggered": false,"sql": "select * from recoverTest","actions": [{"log": {}}]}`
+
+	buf2 := bytes.NewBuffer([]byte(ruleJson))
+	req2, _ := http.NewRequest(http.MethodPost, "http://localhost:8080/rules", buf2)
+	w2 := httptest.NewRecorder()
+	suite.r.ServeHTTP(w2, req2)
+
+	req3, _ := http.NewRequest(http.MethodGet, "http://localhost:8080/rules", bytes.NewBufferString("any"))
+	w3 := httptest.NewRecorder()
+	suite.r.ServeHTTP(w3, req3)
+
+	b, err := io.ReadAll(w3.Result().Body)
+	require.NoError(suite.T(), err)
+	got := make([]map[string]string, 0)
+	require.NoError(suite.T(), json.Unmarshal(b, &got))
+	find := false
+	for _, s := range got {
+		if s["id"] == "recoverTest" {
+			find = true
+			require.Equal(suite.T(), "Stopped: canceled manually.", s["status"])
+		}
+	}
+	require.True(suite.T(), find)
+}
+
 func (suite *RestTestSuite) Test_rulesManageHandler() {
 	connection.InitConnectionManager4Test()
 	// Start rules