From d266d190bd744b7b6f572bf69a42013e21b9be62 Mon Sep 17 00:00:00 2001 From: Bruno Sofiato Date: Thu, 3 Oct 2024 13:03:36 -0300 Subject: [PATCH] Fixed race condition when deleting documents by repoId in ElasticSearch (#32185) Resolves #32184 --------- Signed-off-by: Bruno Sofiato --- .../code/elasticsearch/elasticsearch.go | 28 ++++++++++++++++++- 1 file changed, 27 insertions(+), 1 deletion(-) diff --git a/modules/indexer/code/elasticsearch/elasticsearch.go b/modules/indexer/code/elasticsearch/elasticsearch.go index d64d99433d989..5c01034450be7 100644 --- a/modules/indexer/code/elasticsearch/elasticsearch.go +++ b/modules/indexer/code/elasticsearch/elasticsearch.go @@ -20,6 +20,7 @@ import ( indexer_internal "code.gitea.io/gitea/modules/indexer/internal" inner_elasticsearch "code.gitea.io/gitea/modules/indexer/internal/elasticsearch" "code.gitea.io/gitea/modules/json" + "code.gitea.io/gitea/modules/log" "code.gitea.io/gitea/modules/setting" "code.gitea.io/gitea/modules/timeutil" "code.gitea.io/gitea/modules/typesniffer" @@ -197,8 +198,33 @@ func (b *Indexer) Index(ctx context.Context, repo *repo_model.Repository, sha st return nil } -// Delete deletes indexes by ids +// Delete entries by repoId func (b *Indexer) Delete(ctx context.Context, repoID int64) error { + if err := b.doDelete(ctx, repoID); err != nil { + // Maybe there is a conflict during the delete operation, so we should retry after a refresh + log.Warn("Deletion of entries of repo %v within index %v was erroneus. Trying to refresh index before trying again", repoID, b.inner.VersionedIndexName(), err) + if err := b.refreshIndex(ctx); err != nil { + return err + } + if err := b.doDelete(ctx, repoID); err != nil { + log.Error("Could not delete entries of repo %v within index %v", repoID, b.inner.VersionedIndexName()) + return err + } + } + return nil +} + +func (b *Indexer) refreshIndex(ctx context.Context) error { + if _, err := b.inner.Client.Refresh(b.inner.VersionedIndexName()).Do(ctx); err != nil { + log.Error("Error while trying to refresh index %v", b.inner.VersionedIndexName(), err) + return err + } + + return nil +} + +// Delete entries by repoId +func (b *Indexer) doDelete(ctx context.Context, repoID int64) error { _, err := b.inner.Client.DeleteByQuery(b.inner.VersionedIndexName()). Query(elastic.NewTermsQuery("repo_id", repoID)). Do(ctx)