Analyzer retry: Persist per-table retry timing via available_at and attempts_since_success columns

libs/optimizer/analyzer/src/main/java/com/linkedin/openhouse/optimizer/analyzer/AnalyzerRunner.java

@@ -8,6 +8,7 @@
 import com.linkedin.openhouse.optimizer.repository.TableOperationsHistoryRepository;
 import com.linkedin.openhouse.optimizer.repository.TableOperationsRepository;
 import com.linkedin.openhouse.optimizer.repository.TableStatsRepository;
+import java.time.Instant;
 import java.util.List;
 import java.util.Map;
 import java.util.Optional;
@@ -103,6 +104,19 @@ private List<OperationAnalyzer> selectAnalyzers(Set<OperationTypeDto> operationT
             .collect(Collectors.toList());
   }
 
+  /**
+   * Earliest retry time for {@code operationType} after an outcome observed at {@code from}, using
+   * that analyzer's cadence policy. Empty if no registered analyzer handles the type. Used by the
+   * completion callback to write {@code available_at} on the operation row.
+   */
+  public Optional<Instant> nextAvailableAt(
+      OperationTypeDto operationType, Instant from, boolean succeeded, int attempts) {
+    return analyzers.stream()
+        .filter(a -> a.getOperationType() == operationType)
+        .findFirst()
+        .map(a -> a.nextAvailableAt(from, succeeded, attempts));
+  }
+
   /**
    * Commit-driven worker: evaluate one in-memory table against the given candidate analyzers. The
    * candidates come from the operation-type filter in {@link AnalyzeRequest}; each is additionally
@@ -266,11 +280,57 @@ private boolean analyze(
     if (!analyzer.isEnabled(table)) {
       return false;
     }
+    if (recordExpiryBackoff(analyzer, currentOp)) {
+      // The active op just timed out with no reported outcome; we wrote its retry backoff on this
+      // pass and defer. The gate picks it up once available_at elapses.
+      return false;
+    }
     int consecutiveFailures = countConsecutiveNonSuccess(table, analyzer, currentOp, latestHistory);
     if (!analyzer.shouldSchedule(table, currentOp, latestHistory, consecutiveFailures)) {
       return false;
     }
-    return createPending(analyzer, table);
+    return createPending(analyzer, table, nextAttempts(currentOp, latestHistory));
+  }
+
+  /**
+   * Writes the retry backoff for a dead/missed job: a submitted op that crossed the stale-timeout
+   * without ever reporting a terminal outcome never got an {@code available_at} written by the
+   * completion callback, so its retry would fire immediately on timeout. Write the failure backoff
+   * here — the same interval a reported failure would get — so the dead job backs off like any
+   * other failure. No-op unless the op is timed out and has no {@code available_at} yet. Returns
+   * whether a backoff was written.
+   */
+  private boolean recordExpiryBackoff(
+      OperationAnalyzer analyzer, Optional<TableOperationDto> currentOp) {
+    if (currentOp.isEmpty()) {
+      return false;
+    }
+    TableOperationDto op = currentOp.get();
+    if (op.getAvailableAt() != null || !analyzer.isActiveOpTimedOut(op)) {
+      return false;
+    }
+    Instant availableAt = analyzer.nextAvailableAt(Instant.now(), false, op.getAttempts());
+    operationsRepo.save(op.toRow().toBuilder().availableAt(availableAt).build());
+    log.info(
+        "Operation {} for table {}.{} timed out without reporting; backing off retry until {}",
+        op.getId(),
+        op.getDatabaseName(),
+        op.getTableName(),
+        availableAt);
+    return true;
+  }
+
+  /**
+   * Attempt number for the next PENDING: reset to 1 when the latest run succeeded, otherwise one
+   * past the prior attempt (carried forward on the operation row). Maintained independently of the
+   * history windows so a recorded success can't be miscounted.
+   */
+  private static int nextAttempts(
+      Optional<TableOperationDto> currentOp, Optional<TableOperationsHistoryDto> latestHistory) {
+    boolean lastSucceeded =
+        latestHistory.map(h -> h.getStatus() == HistoryStatusDto.SUCCESS).orElse(false);
+    int prior = lastSucceeded ? 0 : currentOp.map(TableOperationDto::getAttempts).orElse(0);
+    return prior + 1;
   }
 
   /**
@@ -341,9 +401,10 @@ private int countConsecutiveNonSuccess(
    * Persist one PENDING operation, isolating failures so a single bad table never aborts the rest
    * of the pass; the next pass retries it. Returns whether the save succeeded.
    */
-  private boolean createPending(OperationAnalyzer analyzer, TableDto table) {
+  private boolean createPending(OperationAnalyzer analyzer, TableDto table, int attempts) {
     try {
-      operationsRepo.save(TableOperationDto.pending(table, analyzer.getOperationType()).toRow());
+      operationsRepo.save(
+          TableOperationDto.pending(table, analyzer.getOperationType(), attempts).toRow());
       log.debug(
           "Created PENDING {} operation for table {}.{}",
           analyzer.getOperationType(),

libs/optimizer/analyzer/src/main/java/com/linkedin/openhouse/optimizer/analyzer/CadenceBasedOrphanFilesDeletionAnalyzer.java

@@ -5,6 +5,7 @@
 import com.linkedin.openhouse.optimizer.model.TableOperationDto;
 import com.linkedin.openhouse.optimizer.model.TableOperationsHistoryDto;
 import java.time.Duration;
+import java.time.Instant;
 import java.util.Optional;
 import lombok.extern.slf4j.Slf4j;
 import org.springframework.beans.factory.annotation.Autowired;
@@ -102,6 +103,11 @@ public boolean isActiveOpTimedOut(TableOperationDto op) {
     return cadencePolicy.isSubmissionTimedOut(op);
   }
 
+  @Override
+  public Instant nextAvailableAt(Instant from, boolean succeeded, int attempts) {
+    return cadencePolicy.nextAvailableAt(from, succeeded, attempts);
+  }
+
   @Override
   public boolean shouldSchedule(
       TableDto table,

libs/optimizer/analyzer/src/main/java/com/linkedin/openhouse/optimizer/analyzer/CadenceBasedStatsCollectionAnalyzer.java

@@ -5,6 +5,7 @@
 import com.linkedin.openhouse.optimizer.model.TableOperationDto;
 import com.linkedin.openhouse.optimizer.model.TableOperationsHistoryDto;
 import java.time.Duration;
+import java.time.Instant;
 import java.util.Optional;
 import lombok.extern.slf4j.Slf4j;
 import org.springframework.beans.factory.annotation.Autowired;
@@ -100,6 +101,11 @@ public boolean isActiveOpTimedOut(TableOperationDto op) {
     return cadencePolicy.isSubmissionTimedOut(op);
   }
 
+  @Override
+  public Instant nextAvailableAt(Instant from, boolean succeeded, int attempts) {
+    return cadencePolicy.nextAvailableAt(from, succeeded, attempts);
+  }
+
   @Override
   public boolean shouldSchedule(
       TableDto table,

libs/optimizer/analyzer/src/main/java/com/linkedin/openhouse/optimizer/analyzer/CadencePolicy.java

@@ -137,14 +137,33 @@ public boolean shouldSchedule(
       Optional<TableOperationDto> currentOp,
       Optional<TableOperationsHistoryDto> latestHistory,
       int consecutiveFailures) {
-    if (currentOp.isPresent() && isBlockingActiveOp(currentOp.get()).orElse(true)) {
-      return false;
+    if (currentOp.isPresent()) {
+      Instant availableAt = currentOp.get().getAvailableAt();
+      if (availableAt != null) {
+        // A terminal outcome (reported success/failure, or expiry) recorded the next retry time; it
+        // is the authoritative gate. Block until it elapses, then schedule — so a reported failure
+        // retries after its backoff rather than waiting out the stale-timeout, and a dead job backs
+        // off instead of rescheduling immediately.
+        return !Instant.now().isBefore(availableAt);
+      }
+      if (isBlockingActiveOp(currentOp.get()).orElse(true)) {
+        return false;
+      }
     }
     return latestHistory
         .map(entry -> readyAfterHistoryEntry(entry, consecutiveFailures))
         .orElse(true);
   }
 
+  /**
+   * Earliest time to retry after an outcome observed at {@code from}: the success cadence on
+   * success, otherwise the failure backoff for {@code attempts}. This is what callers persist to
+   * {@code available_at} on completion or expiry.
+   */
+  public Instant nextAvailableAt(Instant from, boolean succeeded, int attempts) {
+    return from.plus(succeeded ? successRetryInterval : failureInterval(attempts));
+  }
+
   /**
    * Whether an existing operation row should keep the analyzer from issuing a new recommendation,
    * or {@link Optional#empty()} when the row's status is not one this policy classifies. CANCELED

libs/optimizer/analyzer/src/main/java/com/linkedin/openhouse/optimizer/analyzer/OperationAnalyzer.java

@@ -4,6 +4,7 @@
 import com.linkedin.openhouse.optimizer.model.TableDto;
 import com.linkedin.openhouse.optimizer.model.TableOperationDto;
 import com.linkedin.openhouse.optimizer.model.TableOperationsHistoryDto;
+import java.time.Instant;
 import java.util.Optional;
 
 /**
@@ -55,6 +56,13 @@ default boolean triggersOnCommit(TableDto table) {
    */
   boolean isActiveOpTimedOut(TableOperationDto op);
 
+  /**
+   * Earliest time to retry after an outcome observed at {@code from}: the success cadence on
+   * success, otherwise the failure backoff for {@code attempts}. Persisted to {@code available_at}
+   * on completion or expiry so retry timing no longer depends on the stale-timeout.
+   */
+  Instant nextAvailableAt(Instant from, boolean succeeded, int attempts);
+
   /**
    * Returns {@code true} if a new or refreshed operation record should be upserted.
    *

libs/optimizer/analyzer/src/test/java/com/linkedin/openhouse/optimizer/analyzer/AnalyzerRunnerTest.java

@@ -2,6 +2,7 @@
 
 import static org.assertj.core.api.Assertions.assertThat;
 import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.ArgumentMatchers.anyInt;
 import static org.mockito.ArgumentMatchers.eq;
 import static org.mockito.Mockito.lenient;
 import static org.mockito.Mockito.never;
@@ -146,16 +147,17 @@ void analyze_countsConsecutiveFailures_fromRecentSubmissions_andPassesToShouldSc
   }
 
   @Test
-  void analyze_countsLogicalTimeout_asNonSuccess() {
+  void analyze_timedOutCurrentOpWithoutAvailableAt_recordsBackoff_andDefers() {
     TableStatsRow statsEntity =
         TableStatsRow.builder().tableUuid("uuid-1").databaseName(DB).tableName("tbl1").build();
     TableDto expectedTable = TableDto.fromRow(statsEntity);
+    // A submitted op that crossed the stale timeout with no reported outcome (dead/missed job) and
+    // no available_at yet — the completion callback never ran to write its retry time.
     TableOperationsRow staleOp = submittedOpRow("op-dead", Instant.now().minusSeconds(20_000));
+    Instant backoffUntil = Instant.now().plusSeconds(3_600);
 
     when(statsRepo.find(eq(Optional.of(DB)), eq(Optional.empty()), eq(Optional.empty()), any()))
         .thenReturn(List.of(statsEntity));
-    // Current op is a stale SCHEDULED submission with no history → a logical timeout, no history
-    // row.
     when(operationsRepo.find(
             eq(Optional.of(OFD_DB)),
             eq(Optional.empty()),
@@ -166,9 +168,51 @@ void analyze_countsLogicalTimeout_asNonSuccess() {
             eq(Optional.empty()),
             any()))
         .thenReturn(List.of(staleOp));
+    when(analyzer.isEnabled(expectedTable)).thenReturn(true);
+    when(analyzer.isActiveOpTimedOut(any())).thenReturn(true);
+    when(analyzer.nextAvailableAt(any(Instant.class), eq(false), anyInt()))
+        .thenReturn(backoffUntil);
+
+    runner.analyze(AnalyzeRequest.builder().operationTypes(java.util.Set.of(OFD_TYPE)).build());
+
+    // The dead job's failure backoff is written (so it is not rescheduled immediately) and this
+    // pass defers: no streak walk, no shouldSchedule, no new PENDING row.
+    ArgumentCaptor<TableOperationsRow> captor = ArgumentCaptor.forClass(TableOperationsRow.class);
+    verify(operationsRepo).save(captor.capture());
+    TableOperationsRow savedOp = captor.getValue();
+    assertThat(savedOp.getId()).isEqualTo("op-dead");
+    assertThat(savedOp.getAvailableAt()).isEqualTo(backoffUntil);
+    verify(analyzer, never()).shouldSchedule(any(), any(), any(), anyInt());
+  }
+
+  @Test
+  void analyze_timedOutCurrentOpAlreadyHasAvailableAt_doesNotRewrite() {
+    TableStatsRow statsEntity =
+        TableStatsRow.builder().tableUuid("uuid-1").databaseName(DB).tableName("tbl1").build();
+    TableDto expectedTable = TableDto.fromRow(statsEntity);
+    // Same dead job, but available_at is already set (written on an earlier pass). The gate — not a
+    // rewrite — now governs it, so the expiry-write must be a no-op.
+    TableOperationsRow staleBackedOffOp =
+        submittedOpRow("op-dead", Instant.now().minusSeconds(20_000))
+            .toBuilder()
+            .availableAt(Instant.now().plusSeconds(3_600))
+            .build();
+
+    when(statsRepo.find(eq(Optional.of(DB)), eq(Optional.empty()), eq(Optional.empty()), any()))
+        .thenReturn(List.of(statsEntity));
+    when(operationsRepo.find(
+            eq(Optional.of(OFD_DB)),
+            eq(Optional.empty()),
+            eq(Optional.empty()),
+            eq(Optional.of(DB)),
+            eq(Optional.empty()),
+            eq(Optional.empty()),
+            eq(Optional.empty()),
+            any()))
+        .thenReturn(List.of(staleBackedOffOp));
     when(historyRepo.findLatest(eq(OFD_DB), any())).thenReturn(Collections.emptyList());
     when(operationsRepo.findRecentSubmissions(eq("uuid-1"), eq(OFD_DB), any()))
-        .thenReturn(List.of(staleOp));
+        .thenReturn(List.of(staleBackedOffOp));
     when(historyRepo.findRecent(eq("uuid-1"), eq(OFD_DB), any()))
         .thenReturn(Collections.emptyList());
     when(analyzer.isActiveOpTimedOut(any())).thenReturn(true);
@@ -177,7 +221,9 @@ void analyze_countsLogicalTimeout_asNonSuccess() {
 
     runner.analyze(AnalyzeRequest.builder().operationTypes(java.util.Set.of(OFD_TYPE)).build());
 
-    // The timed-out submission with no history counts as one non-success.
+    // No rewrite (available_at already set); the pass proceeds to the normal gate decision. The
+    // timed-out submission still counts as one non-success in the streak passed to shouldSchedule.
+    verify(operationsRepo, never()).save(any());
     verify(analyzer).shouldSchedule(eq(expectedTable), any(), any(), eq(1));
   }
 
@@ -243,6 +289,86 @@ void analyze_insertsNewRow_forEligibleTableWithNoExistingOp() {
     assertThat(saved.getStatus())
         .isEqualTo(com.linkedin.openhouse.optimizer.db.OperationStatus.PENDING);
     assertThat(saved.getId()).isNotNull();
+    // First attempt since success → counter starts at 1.
+    assertThat(saved.getAttempts()).isEqualTo(1);
+  }
+
+  @Test
+  void analyze_attempts_carriesForwardFromPriorOp() {
+    TableStatsRow statsEntity =
+        TableStatsRow.builder().tableUuid("uuid-1").databaseName(DB).tableName("tbl1").build();
+    TableDto expectedTable = TableDto.fromRow(statsEntity);
+    TableOperationsRow priorOp =
+        TableOperationsRow.builder()
+            .id("prior-op")
+            .status(com.linkedin.openhouse.optimizer.db.OperationStatus.SCHEDULED)
+            .tableUuid("uuid-1")
+            .databaseName(DB)
+            .tableName("tbl1")
+            .operationType(OFD_DB)
+            .createdAt(Instant.now().minusSeconds(10_000))
+            .scheduledAt(Instant.now().minusSeconds(9_000))
+            .attempts(2)
+            .build();
+
+    when(statsRepo.find(eq(Optional.of(DB)), eq(Optional.empty()), eq(Optional.empty()), any()))
+        .thenReturn(List.of(statsEntity));
+    stubOpsFind(List.of(priorOp));
+    when(historyRepo.findLatest(eq(OFD_DB), any())).thenReturn(Collections.emptyList());
+    when(analyzer.isEnabled(expectedTable)).thenReturn(true);
+    when(analyzer.shouldSchedule(eq(expectedTable), any(), any(), anyInt())).thenReturn(true);
+
+    runner.analyze(AnalyzeRequest.builder().operationTypes(java.util.Set.of(OFD_TYPE)).build());
+
+    ArgumentCaptor<TableOperationsRow> captor = ArgumentCaptor.forClass(TableOperationsRow.class);
+    verify(operationsRepo).save(captor.capture());
+    // Prior attempt was 2 and the latest run did not succeed → next attempt is 3.
+    assertThat(captor.getValue().getAttempts()).isEqualTo(3);
+  }
+
+  @Test
+  void analyze_attempts_resetsAfterSuccess() {
+    TableStatsRow statsEntity =
+        TableStatsRow.builder().tableUuid("uuid-1").databaseName(DB).tableName("tbl1").build();
+    TableDto expectedTable = TableDto.fromRow(statsEntity);
+    TableOperationsRow priorOp =
+        TableOperationsRow.builder()
+            .id("prior-op")
+            .status(com.linkedin.openhouse.optimizer.db.OperationStatus.SCHEDULED)
+            .tableUuid("uuid-1")
+            .databaseName(DB)
+            .tableName("tbl1")
+            .operationType(OFD_DB)
+            .createdAt(Instant.now().minusSeconds(10_000))
+            .scheduledAt(Instant.now().minusSeconds(9_000))
+            .attempts(4)
+            .build();
+
+    when(statsRepo.find(eq(Optional.of(DB)), eq(Optional.empty()), eq(Optional.empty()), any()))
+        .thenReturn(List.of(statsEntity));
+    stubOpsFind(List.of(priorOp));
+    // Latest run SUCCEEDED → the streak resets regardless of the prior attempt count.
+    when(historyRepo.findLatest(eq(OFD_DB), any()))
+        .thenReturn(
+            List.of(
+                TableOperationsHistoryRow.builder()
+                    .id("prior-op")
+                    .tableUuid("uuid-1")
+                    .databaseName(DB)
+                    .tableName("tbl1")
+                    .operationType(OFD_DB)
+                    .completedAt(Instant.now().minusSeconds(8_000))
+                    .status(com.linkedin.openhouse.optimizer.db.HistoryStatus.SUCCESS)
+                    .build()));
+    when(analyzer.isEnabled(expectedTable)).thenReturn(true);
+    when(analyzer.shouldSchedule(eq(expectedTable), any(), any(), anyInt())).thenReturn(true);
+
+    runner.analyze(AnalyzeRequest.builder().operationTypes(java.util.Set.of(OFD_TYPE)).build());
+
+    ArgumentCaptor<TableOperationsRow> captor = ArgumentCaptor.forClass(TableOperationsRow.class);
+    verify(operationsRepo).save(captor.capture());
+    // Latest run succeeded → next attempt resets to 1.
+    assertThat(captor.getValue().getAttempts()).isEqualTo(1);
   }
 
   @Test
@@ -444,4 +570,17 @@ void analyzeRequest_withDatabaseFilter_scansOnlyThatDatabase_withoutResolvingAll
     // A supplied database filter short-circuits the all-databases resolution.
     verify(statsRepo, never()).findDistinctDatabaseNames();
   }
+
+  private void stubOpsFind(List<TableOperationsRow> rows) {
+    when(operationsRepo.find(
+            eq(Optional.of(OFD_DB)),
+            eq(Optional.empty()),
+            eq(Optional.empty()),
+            eq(Optional.of(DB)),
+            eq(Optional.empty()),
+            eq(Optional.empty()),
+            eq(Optional.empty()),
+            any()))
+        .thenReturn(rows);
+  }
 }