Fix race with table init channel

The function to mark a table initialized wrongly closed the
channel immediately. This could cause an observer to be woken
up too early (before Commit() has finished) and get an uninitialized
snapshot.

Fix the timing issue by moving the closing into Commit(), next to
the normal watch channel closing. This way the initialized table has
been committed before the init watch channel closes.

This issue was detected by `pkg/k8s/statedb_test.go` in cilium/cilium:

     statedb_test.go:263:
         	Error Trace:	/home/runner/work/cilium/cilium/pkg/k8s/statedb_test.go:263
         	            	/home/runner/work/cilium/cilium/pkg/k8s/statedb_test.go:120
         	Error:      	Not equal:
         	            	expected: "obj1"
         	            	actual  : "garbage"

         	            	Diff:
         	            	--- Expected
         	            	+++ Actual
         	            	@@ -1 +1 @@
         	            	-obj1
         	            	+garbage
         	Test:       	TestStateDBReflector/default

Signed-off-by: Jussi Maki <[email protected]>

Author: joamaki

table.go

@@ -216,10 +216,6 @@ func (t *genTable[Obj]) RegisterInitializer(txn WriteTxn, name string) func(Writ
 						slices.Clone(table.pendingInitializers),
 						func(n string) bool { return n == name },
 					)
-					if !table.initialized && len(table.pendingInitializers) == 0 {
-						close(table.initWatchChan)
-						table.initialized = true
-					}
 				}
 			})
 		}

txn.go

@@ -459,9 +459,18 @@ func (txn *txn) Commit() ReadTxn {
 	root := *db.root.Load()
 	root = slices.Clone(root)
 
+	var initChansToClose []chan struct{}
+
 	// Insert the modified tables into the root tree of tables.
 	for pos, table := range txn.modifiedTables {
 		if table != nil {
+			// Check if tables become initialized. We close the channel only after
+			// we've swapped in the new root so that one cannot get a snapshot of
+			// an uninitialized table after observing the channel closing.
+			if !table.initialized && len(table.pendingInitializers) == 0 {
+				initChansToClose = append(initChansToClose, table.initWatchChan)
+				table.initialized = true
+			}
 			root[pos] = *table
 		}
 	}
@@ -480,6 +489,11 @@ func (txn *txn) Commit() ReadTxn {
 		txn.Notify()
 	}
 
+	// Notify table initializations
+	for _, ch := range initChansToClose {
+		close(ch)
+	}
+
 	txn.db.metrics.WriteTxnDuration(
 		txn.handle,
 		txn.tableNames,