fix(lily): Consider partially completed heights w ERRORs and SKIPs

chain/find.go

@@ -213,31 +213,41 @@ func (g *GapIndexer) findTaskEpochGaps(ctx context.Context) (visor.GapReportList
 		&result,
 		`
 select height, task
-from (
-         select distinct vpr.task, vpr.height, vpr.status_information
-         from visor_processing_reports vpr
-                  right join(
-             select height
-             from (
-                      select task_height_count.height, count(task_height_count.height) cheight
-                      from (
-                               select distinct(task) as task, height
-                               from visor_processing_reports
-                               group by height, task
-                               order by height desc
-                           ) task_height_count
-                      group by task_height_count.height
-                  ) task_count_per_height
-             where task_count_per_height.cheight != ?
-         ) incomplete
-                            on vpr.height = incomplete.height
-         where vpr.status_information != ?
-            or vpr.status_information is null
-     ) incomplete_heights_and_their_completed_tasks
-where height >= ? and height <= ?
+	from (
+	select distinct vpr.task, vpr.height
+	from visor_processing_reports vpr
+	right join(
+		select height, cheight
+		from (
+			select task_height_count.height, status, count(task_height_count.height) cheight
+			from (
+				select task, height, status
+				from visor_processing_reports
+				where height >= ?2 and height <= ?3
+				group by height, task, status
+				order by height desc, task
+			) task_height_count
+			where status = ?5
+			group by task_height_count.height, task_height_count.status
+		) task_count_per_height
+		where task_count_per_height.cheight != ?0
+		and status = ?5
+	) incomplete
+	on vpr.height = incomplete.height
+	where (vpr.status_information != ?1
+	or vpr.status_information is null)
+	and vpr.status = ?5
+	and vpr.height >= ?2 and vpr.height <= ?3
+) incomplete_heights_and_their_completed_tasks
 order by height desc
+;
 `,
-		len(AllTasks), visor.ProcessingStatusInformationNullRound, g.minHeight, g.maxHeight,
+		len(AllTasks), // arg 0
+		visor.ProcessingStatusInformationNullRound, // arg 1
+		g.minHeight, // arg 2
+		g.maxHeight, // arg 3
+		visor.ProcessingStatusError, // arg 4
+		visor.ProcessingStatusOK,  //arg 5
 	)
 	if err != nil {
 		return nil, err

chain/find_test.go

@@ -247,6 +247,7 @@ func TestFind(t *testing.T) {
 		require.NoError(t, err)
 
 		expected := makeGapReportList(tsh2, ActorStatesMinerTask, ActorStatesInitTask)
+		expected = append(expected, makeGapReportList(fakeTipset(t, 8), ActorStatesMinerTask)...)
 		assertGapReportsEqual(t, expected, actual)
 	})
 
@@ -319,6 +320,41 @@ func TestFind(t *testing.T) {
 		expected := makeGapReportList(tsh2, MessagesTask, ActorStatesInitTask)
 		assertGapReportsEqual(t, expected, actual)
 	})
+
+	t.Run("(#775) for each task at epoch 2 there exists an ERROR _and_ an OK on some tasks", func(t *testing.T) {
+		truncateVPR(t, db)
+		initializeVPR(t, db, maxHeight, t.Name(), AllTasks...)
+		// error on some tasks
+		errorEpochTasksVPR(t, db, 2, ActorStatesInitTask, ActorStatesMinerTask)
+
+		strg, err := storage.NewDatabaseFromDB(ctx, db, "public")
+		require.NoError(t, err, "NewDatabaseFromDB")
+
+		actual, err := NewGapIndexer(nil, strg, t.Name(), minHeight, maxHeight, AllTasks).
+			findTaskEpochGaps(ctx)
+		require.NoError(t, err)
+
+		// only expect gaps at height 2
+		expected := makeGapReportList(fakeTipset(t, 2), ActorStatesInitTask, ActorStatesMinerTask)
+		assertGapReportsEqual(t, expected, actual)
+	})
+
+	t.Run("(#773) for each task at epoch 2 there exists a SKIP and an OK", func(t *testing.T) {
+		truncateVPR(t, db)
+		initializeVPR(t, db, maxHeight, t.Name(), AllTasks...)
+		skipEpochSkippedVRP(t, db, 2, AllTasks...)
+		appendOKAtEpochVPR(t, db, 2, AllTasks...)
+
+		strg, err := storage.NewDatabaseFromDB(ctx, db, "public")
+		require.NoError(t, err, "NewDatabaseFromDB")
+
+		actual, err := NewGapIndexer(nil, strg, t.Name(), minHeight, maxHeight, AllTasks).
+			findTaskEpochGaps(ctx)
+		require.NoError(t, err)
+
+		// no gaps should be found since the epoch has OK's for all tasks; the SKIPS are ignored.
+		require.Len(t, actual, 0)
+	})
 }
 
 type assertFields struct {
@@ -421,6 +457,17 @@ func skipEpochSkippedVRP(tb testing.TB, db *pg.DB, epoch int, tasks ...string) {
 	}
 }
 
+func appendOKAtEpochVPR(tb testing.TB, db *pg.DB, epoch int, tasks ...string) {
+	for _, task := range tasks {
+		qsrt := fmt.Sprintf(`
+	insert into public.visor_processing_reports(height, state_root, reporter, task, started_at, completed_at, status, status_information, errors_detected)
+	values(%d, concat(%d, '_state_root'), '%s_appendok', '%s', '2021-01-01 00:00:00.000000 +00:00', '2021-01-21 00:00:00.000000 +00:00', 'OK',null, null);
+			`, epoch, epoch, tb.Name(), task)
+		_, err := db.Exec(qsrt)
+		require.NoError(tb, err)
+	}
+}
+
 func truncateVPR(tb testing.TB, db *pg.DB) {
 	_, err := db.Exec(`TRUNCATE TABLE visor_processing_reports`)
 	require.NoError(tb, err, "visor_processing_report")