mirror of
https://github.com/infiniflow/ragflow.git
synced 2026-11-10 21:32:43 +08:00
fix(agent): retain webhook completion across incremental polls (#20697)
This commit is contained in:
1 parent
808841a91f
commit
d6bfce4473
3 files changed
+77
-7
No files matched your search
@@ -125,16 +125,16 @@ func decodeWebhookID(webhookID string, webhooks map[string]webhookTraceRun) (str
|
||||
}
|
||||
|
||||
// collectWebhookTraceEvents returns only events newer than the supplied cursor.
|
||||
// Completion depends on the stored terminal event, even when it was already read.
|
||||
func collectWebhookTraceEvents(run webhookTraceRun, sinceTS float64, webhookID string) webhookTracePoll {
|
||||
result := newWebhookTracePoll(&webhookID, sinceTS, false)
|
||||
for _, event := range run.Events {
|
||||
eventTS := webhookTraceEventTimestamp(event)
|
||||
if eventTS <= sinceTS {
|
||||
continue
|
||||
}
|
||||
result.Events = append(result.Events, event)
|
||||
if eventTS > result.NextSinceTS {
|
||||
result.NextSinceTS = eventTS
|
||||
if eventTS > sinceTS {
|
||||
result.Events = append(result.Events, event)
|
||||
if eventTS > result.NextSinceTS {
|
||||
result.NextSinceTS = eventTS
|
||||
}
|
||||
}
|
||||
if eventType, _ := event["event"].(string); eventType == "finished" {
|
||||
result.Finished = true
|
||||
|
||||
@@ -14,7 +14,62 @@
|
||||
|
||||
package handler
|
||||
|
||||
import "testing"
|
||||
import (
|
||||
"encoding/json"
|
||||
"testing"
|
||||
)
|
||||
|
||||
// TestPollWebhookTraceKeepsTerminalState verifies completion survives advancing the event cursor.
|
||||
func TestPollWebhookTraceKeepsTerminalState(t *testing.T) {
|
||||
for _, outcome := range []struct {
|
||||
name string
|
||||
success bool
|
||||
}{
|
||||
{"successful run", true},
|
||||
{"failed run", false},
|
||||
} {
|
||||
t.Run(outcome.name, func(t *testing.T) {
|
||||
store := webhookTraceStore{Webhooks: map[string]webhookTraceRun{
|
||||
"50": {StartTS: 50, Events: []map[string]any{
|
||||
{"ts": 51, "event": "message"},
|
||||
{"ts": 52, "event": "finished", "data": map[string]any{"success": outcome.success}},
|
||||
{"ts": 53, "event": "message", "data": map[string]any{"content": "late"}},
|
||||
}},
|
||||
}}
|
||||
raw, err := json.Marshal(store)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
id := encodeWebhookID("50")
|
||||
first, err := pollWebhookTrace(string(raw), 50, id)
|
||||
if err != nil || !first.Finished || len(first.Events) != 2 || first.NextSinceTS != 52 {
|
||||
t.Fatalf("first poll = %+v, error = %v", first, err)
|
||||
}
|
||||
data := first.Events[1]["data"].(map[string]any)
|
||||
if data["success"] != outcome.success {
|
||||
t.Fatalf("terminal outcome = %#v, want success=%t", data, outcome.success)
|
||||
}
|
||||
for _, cursor := range []float64{first.NextSinceTS, 60} {
|
||||
repeated, err := pollWebhookTrace(string(raw), cursor, id)
|
||||
if err != nil || !repeated.Finished || repeated.NextSinceTS != cursor {
|
||||
t.Fatalf("repeat at %v = %+v, error = %v", cursor, repeated, err)
|
||||
}
|
||||
if repeated.WebhookID == nil || *repeated.WebhookID != id || repeated.Events == nil || len(repeated.Events) != 0 {
|
||||
t.Fatalf("repeat at %v = %+v, want same run and no replayed or trailing events", cursor, repeated)
|
||||
}
|
||||
}
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
// TestPollWebhookTraceExhaustedActiveRun does not infer completion merely from an empty event delta.
|
||||
func TestPollWebhookTraceExhaustedActiveRun(t *testing.T) {
|
||||
raw := `{"webhooks":{"50":{"start_ts":50,"events":[{"ts":51,"event":"message"}]}}}`
|
||||
result, err := pollWebhookTrace(raw, 51, encodeWebhookID("50"))
|
||||
if err != nil || result.Finished || result.NextSinceTS != 51 || result.Events == nil || len(result.Events) != 0 {
|
||||
t.Fatalf("active run = %+v, error = %v, want unfinished with no new events", result, err)
|
||||
}
|
||||
}
|
||||
|
||||
// TestPollWebhookTraceMissingRun distinguishes waiting for a trigger from losing an already selected trace.
|
||||
func TestPollWebhookTraceMissingRun(t *testing.T) {
|
||||
|
||||
@@ -371,6 +371,21 @@ func TestGetAgentWebhookLogsPollsTraceIncrementally(t *testing.T) {
|
||||
if !incremental.Data.Finished || incremental.Data.NextSinceTS != finishedTS {
|
||||
t.Fatalf("incremental completion = %+v", incremental.Data)
|
||||
}
|
||||
for _, cursor := range []float64{incremental.Data.NextSinceTS, trailingTS} {
|
||||
repeated := requestWebhookTrace(t, h, "c1", "u1", url.Values{
|
||||
"since_ts": {strconv.FormatFloat(cursor, 'f', -1, 64)},
|
||||
"webhook_id": {*discovery.Data.WebhookID},
|
||||
})
|
||||
if repeated.Code != int(common.CodeSuccess) || repeated.Data == nil || !repeated.Data.Finished {
|
||||
t.Fatalf("repeat at %f = %+v, want completed trace", cursor, repeated)
|
||||
}
|
||||
if repeated.Data.WebhookID == nil || *repeated.Data.WebhookID != *discovery.Data.WebhookID || repeated.Data.NextSinceTS != cursor {
|
||||
t.Fatalf("repeat data = %+v, want unchanged run and cursor", repeated.Data)
|
||||
}
|
||||
if repeated.Data.Events == nil || len(repeated.Data.Events) != 0 {
|
||||
t.Fatalf("repeat events = %#v, want no replayed or trailing events", repeated.Data.Events)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// TestGetAgentWebhookLogsHandlesMissingAndInvalidState covers empty and stale cursors.
|
||||
|
||||
Reference in new issue
Block a user