reconciler: Add 'retriesPending' to WaitUntilReconciled

Extend [Reconciler.WaitUntilReconciled] to also indicate whether retries are
pending for any objects with a revision below or equal to [untilRevision].

The committing of results is split into two: one after normal incremental
processing of pending objects and one after processing retries. This way the
entries that failed to reconcile are pushed to the retry queue and we can check
the low watermark to produce 'retriesPending'.

Signed-off-by: Jussi Maki <jussi.maki@isovalent.com>

Author: joamaki

iterator.go

@@ -6,7 +6,6 @@ package statedb
 import (
 	"fmt"
 	"iter"
-	"runtime"
 	"slices"
 
 	"github.com/cilium/statedb/index"
@@ -253,11 +252,6 @@ func (it *changeIterator[Obj]) nextAny(txn ReadTxn) (iter.Seq2[Change[any], Revi
 }
 
 func (it *changeIterator[Obj]) Close() {
-	runtime.SetFinalizer(it, nil)
-	it.close()
-}
-
-func (it *changeIterator[Obj]) close() {
 	it.iter = nil
 	if it.dt != nil {
 		it.dt.close()

reconciler/incremental.go

@@ -45,39 +45,42 @@ type opResult struct {
 	id       uint64 // the "pending" identifier
 }
 
-func (incr *incremental[Obj]) run(ctx context.Context, txn statedb.ReadTxn, changes iter.Seq2[statedb.Change[Obj], statedb.Revision]) (errs []error, lastRev statedb.Revision, changed bool) {
+func (incr *incremental[Obj]) run(ctx context.Context, txn statedb.ReadTxn, changes iter.Seq2[statedb.Change[Obj], statedb.Revision]) (errs []error, lastRev statedb.Revision, retryLowWatermark statedb.Revision) {
 	// Reconcile new and changed objects using either Operations
 	// or BatchOperations.
 	if incr.config.BatchOperations != nil {
-		lastRev, changed = incr.batch(ctx, txn, changes)
+		lastRev = incr.batch(ctx, txn, changes)
 	} else {
-		lastRev, changed = incr.single(ctx, txn, changes)
+		lastRev = incr.single(ctx, txn, changes)
 	}
 
+	// Commit status updates for new and changed objects.
+	newErrors := incr.commitStatus()
+	clear(incr.results)
+
 	// Process objects that need to be retried that were not cleared.
-	incr.processRetries(ctx, txn)
+	retryLowWatermark = incr.processRetries(ctx, txn)
 
-	// Finally commit the status updates.
-	newErrors := incr.commitStatus()
+	// Finally commit the status updates from retries.
+	newErrors += incr.commitStatus()
 
 	// Since all failures are retried, we can return the errors from the retry
 	// queue which includes both errors occurred in this round and the old
 	// errors.
 	errs = incr.retries.errors()
 	incr.metrics.ReconciliationErrors(incr.moduleID, newErrors, len(errs))
 
-	// Prepare for next round
+	// Prepare for next round.
 	incr.numReconciled = 0
 	clear(incr.results)
 
-	return errs, lastRev, changed
+	return errs, lastRev, retryLowWatermark
 }
 
-func (incr *incremental[Obj]) single(ctx context.Context, txn statedb.ReadTxn, changes iter.Seq2[statedb.Change[Obj], statedb.Revision]) (lastRev statedb.Revision, changed bool) {
+func (incr *incremental[Obj]) single(ctx context.Context, txn statedb.ReadTxn, changes iter.Seq2[statedb.Change[Obj], statedb.Revision]) (lastRev statedb.Revision) {
 	// Iterate in revision order through new and changed objects.
 	for change, rev := range changes {
 		lastRev = rev
-		changed = true
 
 		obj := change.Object
 
@@ -102,14 +105,13 @@ func (incr *incremental[Obj]) single(ctx context.Context, txn statedb.ReadTxn, c
 	return
 }
 
-func (incr *incremental[Obj]) batch(ctx context.Context, txn statedb.ReadTxn, changes iter.Seq2[statedb.Change[Obj], statedb.Revision]) (lastRev statedb.Revision, changed bool) {
+func (incr *incremental[Obj]) batch(ctx context.Context, txn statedb.ReadTxn, changes iter.Seq2[statedb.Change[Obj], statedb.Revision]) (lastRev statedb.Revision) {
 	ops := incr.config.BatchOperations
 	updateBatch := []BatchEntry[Obj]{}
 	deleteBatch := []BatchEntry[Obj]{}
 
 	for change, rev := range changes {
 		lastRev = rev
-		changed = true
 
 		obj := change.Object
 
@@ -152,7 +154,7 @@ func (incr *incremental[Obj]) batch(ctx context.Context, txn statedb.ReadTxn, ch
 		for _, entry := range deleteBatch {
 			if entry.Result != nil {
 				// Delete failed, queue a retry for it.
-				incr.retries.Add(entry.original, entry.Revision, true, entry.Result)
+				incr.retries.Add(entry.original, entry.Revision, entry.Revision, true, entry.Result)
 			}
 		}
 	}
@@ -179,7 +181,7 @@ func (incr *incremental[Obj]) batch(ctx context.Context, txn statedb.ReadTxn, ch
 	return
 }
 
-func (incr *incremental[Obj]) processRetries(ctx context.Context, txn statedb.ReadTxn) {
+func (incr *incremental[Obj]) processRetries(ctx context.Context, txn statedb.ReadTxn) statedb.Revision {
 	now := time.Now()
 	for incr.numReconciled < incr.config.IncrementalRoundSize {
 		item, ok := incr.retries.Top()
@@ -190,6 +192,7 @@ func (incr *incremental[Obj]) processRetries(ctx context.Context, txn statedb.Re
 		incr.processSingle(ctx, txn, item.object.(Obj), item.rev, item.delete)
 		incr.numReconciled++
 	}
+	return incr.retries.LowWatermark()
 }
 
 func (incr *incremental[Obj]) processSingle(ctx context.Context, txn statedb.ReadTxn, obj Obj, rev statedb.Revision, delete bool) {
@@ -204,7 +207,7 @@ func (incr *incremental[Obj]) processSingle(ctx context.Context, txn statedb.Rea
 		err = incr.config.Operations.Delete(ctx, txn, rev, obj)
 		if err != nil {
 			// Deletion failed. Retry again later.
-			incr.retries.Add(obj, rev, true, err)
+			incr.retries.Add(obj, rev, rev, true, err)
 		}
 	} else {
 		// Clone the object so it can be mutated by Update()
@@ -267,7 +270,7 @@ func (incr *incremental[Obj]) commitStatus() (numErrors int) {
 			// Reconciliation of the object had failed and the status was updated
 			// successfully (object had not changed). Queue the retry for the object.
 			newRevision := incr.table.Revision(wtxn)
-			incr.retries.Add(result.original.(Obj), newRevision, false, result.err)
+			incr.retries.Add(result.original.(Obj), newRevision, result.rev, false, result.err)
 		}
 	}
 	return

reconciler/progress.go

@@ -13,9 +13,10 @@ import (
 // progressTracker tracks the highest revision observed as reconciled and
 // allows callers to wait until a target revision is reached.
 type progressTracker struct {
-	mu       sync.Mutex
-	revision statedb.Revision
-	watch    chan struct{}
+	mu                sync.Mutex
+	revision          statedb.Revision
+	retryLowWatermark statedb.Revision
+	watch             chan struct{}
 }
 
 func newProgressTracker() *progressTracker {
@@ -24,29 +25,38 @@ func newProgressTracker() *progressTracker {
 	}
 }
 
-func (p *progressTracker) update(rev statedb.Revision) {
+func (p *progressTracker) update(rev statedb.Revision, retryLowWatermark statedb.Revision) {
 	p.mu.Lock()
+	updated := false
 	if rev > p.revision {
 		p.revision = rev
+		updated = true
+	}
+	if retryLowWatermark != p.retryLowWatermark {
+		p.retryLowWatermark = retryLowWatermark
+		updated = true
+	}
+	if updated {
 		close(p.watch)
 		p.watch = make(chan struct{})
 	}
 	p.mu.Unlock()
 }
 
-func (p *progressTracker) wait(ctx context.Context, rev statedb.Revision) (statedb.Revision, error) {
+func (p *progressTracker) wait(ctx context.Context, rev statedb.Revision) (statedb.Revision, statedb.Revision, error) {
 	for {
 		p.mu.Lock()
 		current := p.revision
+		retryLowWatermark := p.retryLowWatermark
 		watch := p.watch
 		p.mu.Unlock()
 
 		if current >= rev {
-			return current, nil
+			return current, retryLowWatermark, nil
 		}
 		select {
 		case <-ctx.Done():
-			return current, ctx.Err()
+			return current, retryLowWatermark, ctx.Err()
 		case <-watch:
 		}
 	}

reconciler/progress_test.go

@@ -5,10 +5,12 @@ package reconciler_test
 
 import (
 	"context"
+	"errors"
 	"fmt"
 	"iter"
 	"log/slog"
 	"sync"
+	"sync/atomic"
 	"testing"
 	"testing/synctest"
 	"time"
@@ -25,24 +27,25 @@ import (
 
 type waitObject struct {
 	ID     uint64
+	Fail   *atomic.Bool
 	Status reconciler.Status
 }
 
 // TableHeader implements statedb.TableWritable.
-func (w waitObject) TableHeader() []string {
-	return []string{"ID", "Status"}
+func (w *waitObject) TableHeader() []string {
+	return []string{"ID", "Fail", "Status"}
 }
 
 // TableRow implements statedb.TableWritable.
-func (w waitObject) TableRow() []string {
+func (w *waitObject) TableRow() []string {
+	fail := w.Fail.Load()
 	return []string{
 		fmt.Sprintf("%d", w.ID),
+		fmt.Sprintf("%t", fail),
 		w.Status.String(),
 	}
 }
 
-var _ statedb.TableWritable = waitObject{}
-
 func (w *waitObject) Clone() *waitObject {
 	w2 := *w
 	return &w2
@@ -69,14 +72,19 @@ var waitObjectIDIndex = statedb.Index[*waitObject, uint64]{
 type waitOps struct {
 	started     chan struct{}
 	unblock     chan struct{}
-	startedOnce sync.Once
+	markStarted func()
 }
 
 func newWaitOps() *waitOps {
-	return &waitOps{
+	w := &waitOps{
 		started: make(chan struct{}),
 		unblock: make(chan struct{}),
 	}
+
+	w.markStarted = sync.OnceFunc(func() {
+		close(w.started)
+	})
+	return w
 }
 
 // Delete implements reconciler.Operations.
@@ -91,9 +99,10 @@ func (*waitOps) Prune(context.Context, statedb.ReadTxn, iter.Seq2[*waitObject, s
 
 // Update implements reconciler.Operations.
 func (w *waitOps) Update(ctx context.Context, txn statedb.ReadTxn, rev statedb.Revision, obj *waitObject) error {
-	w.startedOnce.Do(func() {
-		close(w.started)
-	})
+	w.markStarted()
+	if obj.Fail.Load() {
+		return errors.New("fail")
+	}
 	select {
 	case <-w.unblock:
 		return nil
@@ -140,6 +149,7 @@ func TestWaitUntilReconciled(t *testing.T) {
 						ops,
 						nil,
 						reconciler.WithoutPruning(),
+						reconciler.WithRetry(10*time.Millisecond, 10*time.Millisecond),
 					)
 					return err
 				}),
@@ -156,32 +166,42 @@ func TestWaitUntilReconciled(t *testing.T) {
 		defer cancel()
 
 		// Won't block if we query with 0 revision.
-		_, err := r.WaitUntilReconciled(waitCtx, 0)
+		_, retryRevision, err := r.WaitUntilReconciled(waitCtx, 0)
 		require.NoError(t, err)
+		require.Zero(t, retryRevision)
 
 		// Insert an object and wait for it to be reconciled.
 		wtxn := db.WriteTxn(table)
 		table.Insert(wtxn, &waitObject{
 			ID:     1,
+			Fail:   new(atomic.Bool),
 			Status: reconciler.StatusPending(),
 		})
 		revision := table.Revision(wtxn)
 		wtxn.Commit()
 
 		type waitResult struct {
-			rev statedb.Revision
-			err error
+			rev           statedb.Revision
+			retryRevision statedb.Revision
+			err           error
 		}
 		done := make(chan waitResult, 1)
 		go func() {
-			rev, err := r.WaitUntilReconciled(waitCtx, revision)
-			done <- waitResult{rev: rev, err: err}
+			rev, retryRevision, err := r.WaitUntilReconciled(waitCtx, revision)
+			done <- waitResult{rev: rev, err: err, retryRevision: retryRevision}
 		}()
 
-		synctest.Wait()
-		select {
-		case <-ops.started:
-		default:
+		started := false
+		for !started {
+			// Advance the fake time
+			time.Sleep(50 * time.Millisecond)
+			select {
+			case <-ops.started:
+				started = true
+			default:
+			}
+		}
+		if !started {
 			t.Fatal("expected update to start")
 		}
 
@@ -192,14 +212,48 @@ func TestWaitUntilReconciled(t *testing.T) {
 		}
 
 		close(ops.unblock)
+
 		synctest.Wait()
 
 		select {
 		case result := <-done:
 			require.NoError(t, result.err)
-			require.Equal(t, revision, result.rev)
+			require.Zero(t, result.retryRevision)
 		default:
 			t.Fatal("expected WaitUntilReconciled to complete")
 		}
+
+		wtxn = db.WriteTxn(table)
+		obj := &waitObject{
+			ID:     2,
+			Fail:   new(atomic.Bool),
+			Status: reconciler.StatusPending(),
+		}
+		obj.Fail.Store(true)
+		table.Insert(wtxn, obj)
+
+		retryRevision = table.Revision(wtxn)
+		wtxn.Commit()
+
+		synctest.Wait()
+
+		origRetryRevision := retryRevision
+		rev, returnedRetryRevision, err := r.WaitUntilReconciled(waitCtx, origRetryRevision)
+		require.NoError(t, err)
+		require.Equal(t, origRetryRevision, rev)
+		require.Equal(t, origRetryRevision, returnedRetryRevision)
+
+		obj.Fail.Store(false)
+		// Advance the fake time enough that a retry has happened.
+		time.Sleep(time.Second)
+
+		obj, newRev, ok := table.Get(db.ReadTxn(), waitObjectIDIndex.Query(2))
+		require.True(t, ok && obj.Status.Kind == reconciler.StatusKindDone)
+
+		rev, returnedRetryRevision, err = r.WaitUntilReconciled(waitCtx, origRetryRevision)
+		require.NoError(t, err)
+		require.Equal(t, newRev, rev)
+		require.Zero(t, returnedRetryRevision)
+
 	})
 }