statedb: Add ChangeIterator.Close method

If we use 'synctest' then we can't close channels outside the synctest bubble.

The Changes() method creates a `*changeIterator` and registers a finalizer for
it that unregisters the delete tracker from the table. As the delete trackers
are stored in a 'part.Map` there's a watch channel that gets closed and this
triggers a panic if this happens in a synctest.

To avoid this issue add a 'Close()' method to the 'ChangeIterator' interface
to allow optionally closing the iterator and avoiding the finalizer.

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

Author: joamaki

derive.go

@@ -74,6 +74,8 @@ func (d derive[In, Out]) loop(ctx context.Context, _ cell.Health) error {
 	if err != nil {
 		return err
 	}
+	defer iter.Close()
+
 	for {
 		wtxn := d.DB.WriteTxn(out)
 		changes, watch := iter.Next(wtxn)

iterator.go

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

observable.go

@@ -32,6 +32,7 @@ func (to *observable[Obj]) Observe(ctx context.Context, next func(Change[Obj]),
 			complete(err)
 			return
 		}
+		defer iter.Close()
 		defer complete(nil)
 
 		for {

reconciler/reconciler.go

@@ -43,6 +43,7 @@ func (r *reconciler[Obj]) reconcileLoop(ctx context.Context, health cell.Health)
 	if err != nil {
 		return fmt.Errorf("watching for changes failed: %w", err)
 	}
+	defer changeIterator.Close()
 
 	tableWatchChan := closedWatchChannel
 

types.go

@@ -110,6 +110,12 @@ type ChangeIterator[Obj any] interface {
 	//
 	// Next will panic if called with a WriteTxn that has locked the target table.
 	Next(ReadTxn) (iter.Seq2[Change[Obj], Revision], <-chan struct{})
+
+	// Close the change iterator. Once all change iterators for a given table are closed
+	// deleted objects for that table are no longer set aside for the change iterators.
+	//
+	// Calling this method is optional as each iterator has a finalizer that closes it.
+	Close()
 }
 
 // RWTable provides methods for modifying the table under a write transaction