diff --git a/internal/engine/elasticsearch/chunk.go b/internal/engine/elasticsearch/chunk.go index 6455d23716..538e1bca8a 100644 --- a/internal/engine/elasticsearch/chunk.go +++ b/internal/engine/elasticsearch/chunk.go @@ -413,6 +413,7 @@ func (e *elasticsearchEngine) updateSingleMemoryMessage(ctx context.Context, ind Index: indexName, DocumentID: messageDocID, Body: bytes.NewReader(body), + Refresh: "wait_for", } res, err := req.Do(ctx, e.client) if err != nil { diff --git a/internal/engine/elasticsearch/chunk_helpers_test.go b/internal/engine/elasticsearch/chunk_helpers_test.go index fbbb71842c..ea000c4869 100644 --- a/internal/engine/elasticsearch/chunk_helpers_test.go +++ b/internal/engine/elasticsearch/chunk_helpers_test.go @@ -1,10 +1,15 @@ package elasticsearch import ( + "context" + "net/http" + "net/http/httptest" "reflect" "strings" "testing" + "github.com/elastic/go-elasticsearch/v8" + "ragflow/internal/engine/types" ) @@ -35,6 +40,37 @@ func TestBuildQueryStringQueryKeepsDocumentFieldsUnchanged(t *testing.T) { assertEqual(t, queryString["fields"], []string{"name^10"}) } +func TestUpdateSingleMemoryMessageWaitsForRefresh(t *testing.T) { + var gotRefresh string + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if r.URL.Path != "/memory_tenant/_update/memory-1_42" { + t.Errorf("path=%s, want /memory_tenant/_update/memory-1_42", r.URL.Path) + http.Error(w, "unexpected request path", http.StatusNotFound) + return + } + gotRefresh = r.URL.Query().Get("refresh") + w.Header().Set("X-Elastic-Product", "Elasticsearch") + w.Header().Set("Content-Type", "application/json") + _, _ = w.Write([]byte(`{"result":"updated"}`)) + })) + defer server.Close() + + client, err := elasticsearch.NewClient(elasticsearch.Config{ + Addresses: []string{server.URL}, + }) + if err != nil { + t.Fatalf("new elasticsearch client: %v", err) + } + + engine := &elasticsearchEngine{client: client} + if err := engine.updateSingleMemoryMessage(context.Background(), "memory_tenant", "memory-1_42", map[string]interface{}{"forget_at": "2026-07-27 10:00:00"}); err != nil { + t.Fatalf("updateSingleMemoryMessage: %v", err) + } + if gotRefresh != "wait_for" { + t.Fatalf("refresh=%q, want wait_for", gotRefresh) + } +} + func TestElasticsearchGetFieldsFiltersAndUsesIDFallback(t *testing.T) { engine := &elasticsearchEngine{} chunks := []map[string]interface{}{