Fixed race condition when deleting documents by repoId in ElasticSearch (#32185)

Resolves #32184

---------

Signed-off-by: Bruno Sofiato <bruno.sofiato@gmail.com>

diff --git a/modules/indexer/code/elasticsearch/elasticsearch.go b/modules/indexer/code/elasticsearch/elasticsearch.go
index d64d99433d989685fba2777845107e272611bb0d..5c01034450be7987a69dce82160c3e59661a39d7 100644 (file)
--- 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)