PLM-131: Failed index reports and recreate (#111)

Co-authored-by: github-actions[bot] <41898282+github-actions[bot]@users.noreply.github.com>
Co-authored-by: Copilot <[email protected]>

Author: 3 people

plm/catalog.go

@@ -99,10 +99,11 @@ type collectionCatalog struct {
 type indexCatalogEntry struct {
 	*topo.IndexSpecification
 	Incomplete bool `bson:"incomplete"`
+	Failed     bool `bson:"failed"`
 }
 
-func (i indexCatalogEntry) Ready() bool {
-	return !i.Incomplete
+func (i indexCatalogEntry) Unsuccessful() bool {
+	return i.Failed || i.Incomplete
 }
 
 // NewCatalog creates a new Catalog.
@@ -401,10 +402,13 @@ func (c *Catalog) CreateIndexes(
 
 	successfulIdxs := make([]indexCatalogEntry, 0, len(processedIdxs))
 	successfulIdxNames := make([]string, 0, len(processedIdxs))
+
+	failedIdxs := make([]*topo.IndexSpecification, 0, len(processedIdxs))
 	var idxErrors []error
 
 	for _, idx := range indexes {
 		if err := processedIdxs[idx.Name]; err != nil {
+			failedIdxs = append(failedIdxs, idx)
 			idxErrors = append(idxErrors, errors.Wrap(err, "create index: "+idx.Name))
 
 			continue
@@ -418,6 +422,8 @@ func (c *Catalog) CreateIndexes(
 	c.addIndexesToCatalog(ctx, db, coll, successfulIdxs)
 
 	if len(idxErrors) > 0 {
+		c.AddFailedIndexes(ctx, db, coll, failedIdxs)
+
 		lg.Errorf(errors.Join(idxErrors...),
 			"One or more indexes failed to create on %s.%s", db, coll)
 	}
@@ -457,6 +463,35 @@ func (c *Catalog) AddIncompleteIndexes(
 	c.addIndexesToCatalog(ctx, db, coll, indexEntries)
 }
 
+// AddFailedIndexes adds indexes in the catalog that failed to create on the target cluster.
+// The indexes have set [indexCatalogEntry.Failed] flag.
+func (c *Catalog) AddFailedIndexes(
+	ctx context.Context,
+	db string,
+	coll string,
+	indexes []*topo.IndexSpecification,
+) {
+	lg := log.Ctx(ctx)
+
+	if len(indexes) == 0 {
+		lg.Error(nil, "No failed indexes to add")
+
+		return
+	}
+
+	indexEntries := make([]indexCatalogEntry, len(indexes))
+	for i, index := range indexes {
+		indexEntries[i] = indexCatalogEntry{
+			IndexSpecification: index,
+			Failed:             true,
+		}
+
+		lg.Tracef("Added failed index %q for %s.%s to catalog", index.Name, db, coll)
+	}
+
+	c.addIndexesToCatalog(ctx, db, coll, indexEntries)
+}
+
 // ModifyCappedCollection modifies a capped collection in the target MongoDB.
 func (c *Catalog) ModifyCappedCollection(
 	ctx context.Context,
@@ -725,12 +760,13 @@ func (c *Catalog) Finalize(ctx context.Context) error {
 
 	var idxErrors []error
 
+	foundUnsuccessfulIdx := false
+
 	for db, colls := range c.Databases {
 		for coll, collEntry := range colls.Collections {
 			for _, index := range collEntry.Indexes {
-				if !index.Ready() {
-					lg.Warnf("Index %s on %s.%s was incomplete during replication, skipping it",
-						index.Name, db, coll)
+				if index.Unsuccessful() {
+					foundUnsuccessfulIdx = true
 
 					continue
 				}
@@ -804,13 +840,61 @@ func (c *Catalog) Finalize(ctx context.Context) error {
 		}
 	}
 
+	if foundUnsuccessfulIdx {
+		c.finalizeUnsuccessfulIndexes(ctx)
+	}
+
 	if len(idxErrors) > 0 {
 		lg.Errorf(errors.Join(idxErrors...), "Finalize indexes")
 	}
 
 	return nil
 }
 
+// finalizeUnsuccessfulIndexes finalizes indexes that were unsuccessful
+// during replication, failed or incomplete.
+func (c *Catalog) finalizeUnsuccessfulIndexes(ctx context.Context) {
+	lg := log.Ctx(ctx)
+	lg.Info("Finalizing unsuccessful indexes")
+
+	for db, colls := range c.Databases {
+		for coll, collEntry := range colls.Collections {
+			for _, index := range collEntry.Indexes {
+				if !index.Unsuccessful() {
+					continue // skip successful indexes
+				}
+
+				if index.Incomplete {
+					lg.Infof("Index %s on %s.%s was incomplete during replication, trying to create it",
+						index.Name, db, coll)
+				}
+
+				if index.Failed {
+					lg.Infof("Index %s on %s.%s failed to create during replication, trying to recreate it",
+						index.Name, db, coll)
+				}
+
+				err := runWithRetry(ctx, func(ctx context.Context) error {
+					return c.target.Database(db).RunCommand(ctx, bson.D{
+						{"createIndexes", coll},
+						{"indexes", bson.A{index.IndexSpecification}},
+					}).Err()
+				})
+				if err != nil {
+					lg.Warnf("Failed to recreate unsuccessful index %s on %s.%s: %v",
+						index.Name, db, coll, err)
+
+					continue
+				}
+
+				lg.Infof("Recreated index %s on %s.%s", index.Name, db, coll)
+
+				c.addIndexesToCatalog(ctx, db, coll, []indexCatalogEntry{{IndexSpecification: index.IndexSpecification}})
+			}
+		}
+	}
+}
+
 // doModifyIndexOption modifies an index property in the target MongoDB.
 func (c *Catalog) doModifyIndexOption(
 	ctx context.Context,
@@ -888,9 +972,6 @@ func (c *Catalog) addIndexesToCatalog(
 
 		for i, catIndex := range collCat.Indexes {
 			if catIndex.Name == index.Name {
-				lg.Warnf("add indexes: index %q already exists in %q namespace",
-					index.Name, db+"."+coll)
-
 				collCat.Indexes[i] = index
 				found = true
 

plm/plm.go

@@ -620,7 +620,7 @@ func (ml *PLM) Finalize(ctx context.Context, options FinalizeOptions) error {
 		return errors.New("initial sync is not completed")
 	}
 
-	lg := log.Ctx(ctx)
+	lg := log.New("finalize")
 	lg.Info("Starting Finalization")
 
 	if status.Repl.IsRunning() {
@@ -663,7 +663,7 @@ func (ml *PLM) Finalize(ctx context.Context, options FinalizeOptions) error {
 		go ml.onStateChanged(StateFinalized)
 	}()
 
-	log.New("plm").Info("Finalizing")
+	lg.Info("Finalizing")
 
 	go ml.onStateChanged(StateFinalizing)
 

tests/test_indexes.py

@@ -32,6 +32,15 @@ def test_create_unique(t: Testing, phase: Runner.Phase):
     t.compare_all()
 
 
+@pytest.mark.parametrize("phase", [Runner.Phase.APPLY, Runner.Phase.CLONE])
+def test_create_non_unique_with_the_same_fields_as_unique(t: Testing, phase: Runner.Phase):
+    with t.run(phase):
+        t.source["db_1"]["coll_1"].create_index({"i": 1}, unique=True, name="unique_idx")
+        t.source["db_1"]["coll_1"].create_index({"i": 1}, name="non_unique_idx")
+
+    t.compare_all()
+
+
 @pytest.mark.parametrize("phase", [Runner.Phase.APPLY, Runner.Phase.CLONE])
 def test_create_prepare_unique(t: Testing, phase: Runner.Phase):
     with t.run(phase):
@@ -500,10 +509,7 @@ def test_continue_creating_indexes_if_some_fail(t: Testing, phase: Runner.Phase)
             name="idx_3",
         )
 
-    target_idx_count = len(t.target["db_1"]["coll_1"].index_information())
-    source_idx_count = len(t.source["db_1"]["coll_1"].index_information())
-
-    assert source_idx_count - 1 == target_idx_count
+    t.compare_all()
 
 
 def test_plm_95_drop_index_for_non_existing_namespace(t: Testing):