pass debounce options as a parameter

Take *DebounceOptions on HandleSubscription instead of exposing a second
entry point, so the gateway-api call site keeps its handler inline rather
than hoisting it into a variable to feed two branches. Other callers pass
nil.

Signed-off-by: Huabing (Robin) Zhao <zhaohuabing@gmail.com>
Signed-off-by: Huabing (Robin) Zhao <huabing@tetrate.io>

Author: zhaohuabing

internal/cmd/server.go

@@ -63,6 +63,7 @@ func GetServerCommand(asyncErrHandler func(string, error)) *cobra.Command {
 						asyncErrHandler(update.Key, update.Value)
 					}
 				},
+				nil,
 			)
 
 			hook := func(c context.Context, cfg *config.Server) error {

internal/gatewayapi/runner/runner.go

@@ -184,6 +184,7 @@ func (r *Runner) translateFromSubscription(ctx context.Context, c <-chan watchab
 				}
 			}
 		},
+		nil,
 	)
 	r.Logger.Info("subscriber shutting down")
 }

internal/globalratelimit/runner/runner.go

@@ -147,6 +147,7 @@ func (r *Runner) updateProxyInfraFromSubscription(ctx context.Context, sub <-cha
 				}
 			}
 		},
+		nil,
 	)
 	select {
 	case <-ctx.Done():

internal/infrastructure/runner/runner.go

@@ -93,33 +93,13 @@ func handleWithCrashRecovery[K comparable, V any](
 // This is better than simply iterating over snapshot.Updates because
 // it handles the case where the watchable.Map already contains
 // entries before .Subscribe is called.
-func HandleSubscription[K comparable, V any](l logging.Logger,
-	meta Metadata,
-	subscription <-chan watchable.Snapshot[K, V],
-	handle func(updateFunc Update[K, V], errChans chan error),
-) {
-	handleSubscription(l, meta, subscription, handle, nil)
-}
-
-// HandleSubscriptionWithDebounce is HandleSubscription with time-based coalescing:
-// bursts of updates are merged before being handed off, which bounds how often the
-// handler runs when the upstream source is churning.
 //
-// A pending batch is flushed when either no new update has arrived for After, or the
-// batch has been held for Max. The former keeps propagation fast for isolated changes;
-// the latter caps how long propagation can be delayed while changes keep arriving.
-func HandleSubscriptionWithDebounce[K comparable, V any](l logging.Logger,
-	meta Metadata,
-	subscription <-chan watchable.Snapshot[K, V],
-	handle func(updateFunc Update[K, V], errChans chan error),
-	debounce DebounceOptions,
-) {
-	handleSubscription(l, meta, subscription, handle, &debounce)
-}
-
-// handleSubscription implements both entry points. A nil debounce selects the
-// undebounced loop, in which every delivered snapshot is handled immediately.
-func handleSubscription[K comparable, V any](l logging.Logger,
+// A non-nil debounce merges bursts of updates before handing them off, which bounds
+// how often handle runs when the upstream source is churning. A pending batch is
+// flushed when either no new update has arrived for After, or the batch has been held
+// for Max. The former keeps propagation fast for isolated changes; the latter caps how
+// long propagation can be delayed while changes keep arriving.
+func HandleSubscription[K comparable, V any](l logging.Logger,
 	meta Metadata,
 	subscription <-chan watchable.Snapshot[K, V],
 	handle func(updateFunc Update[K, V], errChans chan error),

internal/message/watchutil.go

@@ -31,6 +31,7 @@ func TestHandleSubscriptionAlreadyClosed(t *testing.T) {
 		message.Metadata{Runner: "demo", Message: "demo"},
 		ch,
 		func(_ message.Update[string, any], _ chan error) { calls++ },
+		nil,
 	)
 	require.Equal(t, 0, calls)
 }
@@ -60,6 +61,7 @@ func TestPanicInSubscriptionHandler(t *testing.T) {
 			numCalls++
 			panic("oops " + update.Key)
 		},
+		nil,
 	)
 	require.Equal(t, 2, numCalls)
 }
@@ -93,6 +95,7 @@ func TestHandleSubscriptionAlreadyInitialized(t *testing.T) {
 				storeCalls++
 			}
 		},
+		nil,
 	)
 	require.LessOrEqual(t, storeCalls, 2) // updates can be coalesced
 	require.Equal(t, 1, deleteCalls)
@@ -283,7 +286,9 @@ func TestControllerResourceUpdate(t *testing.T) {
 					if u.Key == "end" {
 						m.GatewayAPIResources.Close()
 					}
-				})
+				},
+				nil,
+			)
 			if tc.updates > 1 {
 				require.LessOrEqual(t, updates, tc.updates) // Updates can be coalesced
 			} else {
@@ -315,14 +320,14 @@ func TestHandleSubscriptionDebounceQuietPeriod(t *testing.T) {
 	}()
 
 	var updates []any
-	message.HandleSubscriptionWithDebounce(
+	message.HandleSubscription(
 		logging.NewLogger(t.Output(), egv1a1.DefaultEnvoyGatewayLogging()),
 		message.Metadata{Runner: "demo", Message: "demo"},
 		sub,
 		func(update message.Update[string, any], _ chan error) {
 			updates = append(updates, update.Value)
 		},
-		message.DebounceOptions{After: 200 * time.Millisecond, Max: 10 * time.Second},
+		&message.DebounceOptions{After: 200 * time.Millisecond, Max: 10 * time.Second},
 	)
 
 	// Without debouncing this burst yields up to 10 calls; with it the batch is
@@ -363,7 +368,7 @@ func TestHandleSubscriptionDebounceMaxDelay(t *testing.T) {
 	start := time.Now()
 
 	var updates int
-	message.HandleSubscriptionWithDebounce(
+	message.HandleSubscription(
 		logging.NewLogger(t.Output(), egv1a1.DefaultEnvoyGatewayLogging()),
 		message.Metadata{Runner: "demo", Message: "demo"},
 		m.Subscribe(context.Background()),
@@ -375,7 +380,7 @@ func TestHandleSubscriptionDebounceMaxDelay(t *testing.T) {
 			}
 		},
 		// Quiet period far longer than the store interval, so only max can flush.
-		message.DebounceOptions{After: time.Hour, Max: 200 * time.Millisecond},
+		&message.DebounceOptions{After: time.Hour, Max: 200 * time.Millisecond},
 	)
 
 	require.NotZero(t, updates, "max delay should have forced at least one flush")
@@ -399,14 +404,14 @@ func TestHandleSubscriptionDebounceFlushesOnClose(t *testing.T) {
 	}()
 
 	var updates []message.Update[string, any]
-	message.HandleSubscriptionWithDebounce(
+	message.HandleSubscription(
 		logging.NewLogger(t.Output(), egv1a1.DefaultEnvoyGatewayLogging()),
 		message.Metadata{Runner: "demo", Message: "demo"},
 		m.Subscribe(context.Background()),
 		func(update message.Update[string, any], _ chan error) {
 			updates = append(updates, update)
 		},
-		message.DebounceOptions{After: 30 * time.Second, Max: time.Hour},
+		&message.DebounceOptions{After: 30 * time.Second, Max: time.Hour},
 	)
 
 	require.Len(t, updates, 1)
@@ -437,7 +442,7 @@ func TestHandleSubscriptionDebounceCoalescesAcrossKeys(t *testing.T) {
 
 	got := map[string]any{}
 	deletes := map[string]bool{}
-	message.HandleSubscriptionWithDebounce(
+	message.HandleSubscription(
 		logging.NewLogger(t.Output(), egv1a1.DefaultEnvoyGatewayLogging()),
 		message.Metadata{Runner: "demo", Message: "demo"},
 		sub,
@@ -448,7 +453,7 @@ func TestHandleSubscriptionDebounceCoalescesAcrossKeys(t *testing.T) {
 			}
 			got[update.Key] = update.Value
 		},
-		message.DebounceOptions{After: 150 * time.Millisecond, Max: 10 * time.Second},
+		&message.DebounceOptions{After: 150 * time.Millisecond, Max: 10 * time.Second},
 	)
 
 	// Last write wins per key, and the trailing delete for "baz" survives.

internal/message/watchutil_test.go

@@ -59,6 +59,7 @@ func (r *gatewayAPIReconciler) updateStatusFromSubscriptions(ctx context.Context
 					}),
 				})
 			},
+			nil,
 		)
 		r.log.Info("gatewayclass status subscriber shutting down")
 	}()
@@ -84,6 +85,7 @@ func (r *gatewayAPIReconciler) updateStatusFromSubscriptions(ctx context.Context
 				gtw.Status = *update.Value
 				r.updateStatusForGateway(ctx, gtw)
 			},
+			nil,
 		)
 		r.log.Info("gateway status subscriber shutting down")
 	}()
@@ -126,6 +128,7 @@ func (r *gatewayAPIReconciler) updateStatusFromSubscriptions(ctx context.Context
 					}),
 				})
 			},
+			nil,
 		)
 		r.log.Info("httpRoute status subscriber shutting down")
 	}()
@@ -168,6 +171,7 @@ func (r *gatewayAPIReconciler) updateStatusFromSubscriptions(ctx context.Context
 					}),
 				})
 			},
+			nil,
 		)
 		r.log.Info("grpcRoute status subscriber shutting down")
 	}()
@@ -210,6 +214,7 @@ func (r *gatewayAPIReconciler) updateStatusFromSubscriptions(ctx context.Context
 					}),
 				})
 			},
+			nil,
 		)
 		r.log.Info("tlsRoute status subscriber shutting down")
 	}()
@@ -252,6 +257,7 @@ func (r *gatewayAPIReconciler) updateStatusFromSubscriptions(ctx context.Context
 					}),
 				})
 			},
+			nil,
 		)
 		r.log.Info("tcpRoute status subscriber shutting down")
 	}()
@@ -294,6 +300,7 @@ func (r *gatewayAPIReconciler) updateStatusFromSubscriptions(ctx context.Context
 					}),
 				})
 			},
+			nil,
 		)
 		r.log.Info("udpRoute status subscriber shutting down")
 	}()
@@ -331,6 +338,7 @@ func (r *gatewayAPIReconciler) updateStatusFromSubscriptions(ctx context.Context
 					}),
 				})
 			},
+			nil,
 		)
 		r.log.Info("listenerSet status subscriber shutting down")
 	}()
@@ -369,6 +377,7 @@ func (r *gatewayAPIReconciler) updateStatusFromSubscriptions(ctx context.Context
 					}),
 				})
 			},
+			nil,
 		)
 		r.log.Info("envoyPatchPolicy status subscriber shutting down")
 	}()
@@ -407,6 +416,7 @@ func (r *gatewayAPIReconciler) updateStatusFromSubscriptions(ctx context.Context
 					}),
 				})
 			},
+			nil,
 		)
 		r.log.Info("clientTrafficPolicy status subscriber shutting down")
 	}()
@@ -445,6 +455,7 @@ func (r *gatewayAPIReconciler) updateStatusFromSubscriptions(ctx context.Context
 					}),
 				})
 			},
+			nil,
 		)
 		r.log.Info("backendTrafficPolicy status subscriber shutting down")
 	}()
@@ -483,6 +494,7 @@ func (r *gatewayAPIReconciler) updateStatusFromSubscriptions(ctx context.Context
 					}),
 				})
 			},
+			nil,
 		)
 		r.log.Info("securityPolicy status subscriber shutting down")
 	}()
@@ -524,6 +536,7 @@ func (r *gatewayAPIReconciler) updateStatusFromSubscriptions(ctx context.Context
 					}),
 				})
 			},
+			nil,
 		)
 		r.log.Info("backendTlsPolicy status subscriber shutting down")
 	}()
@@ -562,6 +575,7 @@ func (r *gatewayAPIReconciler) updateStatusFromSubscriptions(ctx context.Context
 					}),
 				})
 			},
+			nil,
 		)
 		r.log.Info("envoyExtensionPolicy status subscriber shutting down")
 	}()
@@ -600,6 +614,7 @@ func (r *gatewayAPIReconciler) updateStatusFromSubscriptions(ctx context.Context
 					}),
 				})
 			},
+			nil,
 		)
 		r.log.Info("backend status subscriber shutting down")
 	}()
@@ -644,6 +659,7 @@ func (r *gatewayAPIReconciler) updateStatusFromSubscriptions(ctx context.Context
 						}),
 					})
 				},
+				nil,
 			)
 			r.log.Info("extensionServerPolicies status subscriber shutting down")
 		}()
@@ -685,6 +701,7 @@ func (r *gatewayAPIReconciler) updateStatusFromSubscriptions(ctx context.Context
 					}),
 				})
 			},
+			nil,
 		)
 		r.log.Info("envoyProxy status subscriber shutting down")
 	}()

internal/provider/kubernetes/status.go

@@ -395,6 +395,7 @@ func (r *Runner) translateFromSubscription(sub <-chan watchable.Snapshot[string,
 				}
 			}
 		},
+		nil,
 	)
 	r.Logger.Info("subscriber shutting down")
 }