@@ -312,64 +312,93 @@ func TestInFlightLoadProducer_DumpStateCapsEndpoints(t *testing.T) {
312312 require .Equal (t , int64 (104 ), state .Endpoints [0 ].Requests )
313313}
314314
315- // TestInFlightLoadProducer_FlapDoesNotUnderflow asserts that an in-flight request's release lands
316- // on the exact counter instance it incremented, even after the endpoint is deleted and recreated
317- // under the same NamespacedName. A release from a request that predates the flap hits the orphaned
318- // counter, leaving the live counter accurate and never negative.
315+ // TestInFlightLoadProducer_FlapDoesNotUnderflow asserts that a request's release lands on the exact
316+ // counter instance it incremented, even after the endpoint is deleted and recreated under the same
317+ // NamespacedName. A release that predates the flap hits the orphaned counter, leaving the live
318+ // counter accurate and never negative. Cases cover the request and token counters across both the
319+ // EndOfStream eviction path (OnEvicted) and the StartOfStream early-release path (releaseTokensEarly).
319320func TestInFlightLoadProducer_FlapDoesNotUnderflow (t * testing.T ) {
320321 t .Parallel ()
321322
322- producer := newTestProducer (t )
323- ctx := context .Background ()
324- endpointName := "flap-endpoint"
325- endpointID := fullEndpointName (endpointName )
326-
327- // E joins the endpoint set. The same Endpoint object is reused for the
328- // matching delete so the registeredEndpoints stale-delete guard allows
329- // cleanup (mirrors the datalayer's same-pointer add/delete contract).
330- epA := newStubSchedulingEndpoint (endpointName )
331- require .NoError (t , producer .Extract (ctx , datalayer.EndpointEvent {
332- Type : datalayer .EventAddOrUpdate ,
333- Endpoint : epA ,
334- }))
335-
336- // Request A is routed to endpoint E.
337- reqA := makeTokenRequest ("req-A" , 4 )
338- resA := makeSchedulingResult (endpointName )
339- producer .PreRequest (ctx , reqA , resA )
340- require .Equal (t , int64 (1 ), producer .requestTracker .get (endpointID ))
323+ requests := func (p * InFlightLoadProducer , id string ) int64 { return p .requestTracker .get (id ) }
324+ tokens := func (p * InFlightLoadProducer , id string ) int64 { return p .tokenTracker .get (id ) }
325+
326+ tests := []struct {
327+ name string
328+ addEstimatedOutputTokens bool
329+ release requestcontrol.Response // chunk that triggers the release under test
330+ read func (* InFlightLoadProducer , string ) int64
331+ inputA , inputB int // input tokens for requests A and B
332+ liveAfterA , liveAfterB int64 // live counter after A's, then B's, PreRequest
333+ liveAfterReleaseA int64 // live counter after A releases, while B is still in flight
334+ }{
335+ {
336+ name : "request counter, EndOfStream eviction" ,
337+ release : requestcontrol.Response {EndOfStream : true },
338+ read : requests ,
339+ inputA : 4 , inputB : 4 ,
340+ liveAfterA : 1 , liveAfterB : 1 , liveAfterReleaseA : 1 ,
341+ },
342+ {
343+ name : "token counter, EndOfStream eviction" ,
344+ addEstimatedOutputTokens : true , // 4 input + 6 estimated output = 10 tokens per request
345+ release : requestcontrol.Response {EndOfStream : true },
346+ read : tokens ,
347+ inputA : 4 , inputB : 4 ,
348+ liveAfterA : 10 , liveAfterB : 10 , liveAfterReleaseA : 10 ,
349+ },
350+ {
351+ name : "token counter, StartOfStream early release" ,
352+ release : requestcontrol.Response {StartOfStream : true }, // output excluded: tokens == input
353+ read : tokens ,
354+ inputA : 4 , inputB : 6 ,
355+ liveAfterA : 4 , liveAfterB : 6 , liveAfterReleaseA : 6 ,
356+ },
357+ }
341358
342- // E flaps: it leaves the endpoint set (NotReady) while A is still in flight,
343- // dropping its counter.
344- require .NoError (t , producer .Extract (ctx , datalayer.EndpointEvent {
345- Type : datalayer .EventDelete ,
346- Endpoint : epA ,
347- }))
348- require .Equal (t , int64 (0 ), producer .requestTracker .get (endpointID ), "counter dropped on delete" )
359+ const endpointName = "flap-endpoint"
360+ extract := func (t * testing.T , p * InFlightLoadProducer , eventType datalayer.EventType , ep datalayer.Endpoint ) {
361+ require .NoError (t , p .Extract (context .Background (), datalayer.EndpointEvent {Type : eventType , Endpoint : ep }))
362+ }
349363
350- // E rejoins with the same NamespacedName but a new Endpoint object; request
351- // B's PreRequest recreates a fresh counter instance under the same key.
352- epB := newStubSchedulingEndpoint (endpointName )
353- require .NoError (t , producer .Extract (ctx , datalayer.EndpointEvent {
354- Type : datalayer .EventAddOrUpdate ,
355- Endpoint : epB ,
356- }))
357- reqB := makeTokenRequest ("req-B" , 4 )
358- resB := makeSchedulingResult (endpointName )
359- producer .PreRequest (ctx , reqB , resB )
360- require .Equal (t , int64 (1 ), producer .requestTracker .get (endpointID ), "B recreated the counter" )
361-
362- // A completes. Its release must hit the orphaned counter, not B's live one, which still holds B.
363- reqA .SchedulingResult = resA
364- producer .ResponseBody (ctx , reqA , & requestcontrol.Response {EndOfStream : true }, nil )
365- require .Equal (t , int64 (1 ), producer .requestTracker .get (endpointID ),
366- "A's release must not discount B's live counter" )
367-
368- // B completes. The live counter settles at 0 and never underflows.
369- reqB .SchedulingResult = resB
370- producer .ResponseBody (ctx , reqB , & requestcontrol.Response {EndOfStream : true }, nil )
371- require .Equal (t , int64 (0 ), producer .requestTracker .get (endpointID ),
372- "counter must settle at 0, never negative" )
364+ for _ , tc := range tests {
365+ t .Run (tc .name , func (t * testing.T ) {
366+ producer := newTestProducer (t )
367+ producer .addEstimatedOutputTokens = tc .addEstimatedOutputTokens
368+ ctx := context .Background ()
369+ endpointID := fullEndpointName (endpointName )
370+
371+ // E joins. The same Endpoint object is reused for its delete so the registeredEndpoints
372+ // guard allows cleanup (the datalayer's same-pointer add/delete contract).
373+ epA := newStubSchedulingEndpoint (endpointName )
374+ extract (t , producer , datalayer .EventAddOrUpdate , epA )
375+ reqA := makeTokenRequest ("req-A" , tc .inputA )
376+ resA := makeSchedulingResult (endpointName )
377+ producer .PreRequest (ctx , reqA , resA )
378+ require .Equal (t , tc .liveAfterA , tc .read (producer , endpointID ))
379+
380+ // E flaps and rejoins under the same NamespacedName; B recreates a fresh counter instance.
381+ extract (t , producer , datalayer .EventDelete , epA )
382+ require .Equal (t , int64 (0 ), tc .read (producer , endpointID ), "counter dropped on delete" )
383+ epB := newStubSchedulingEndpoint (endpointName )
384+ extract (t , producer , datalayer .EventAddOrUpdate , epB )
385+ reqB := makeTokenRequest ("req-B" , tc .inputB )
386+ resB := makeSchedulingResult (endpointName )
387+ producer .PreRequest (ctx , reqB , resB )
388+ require .Equal (t , tc .liveAfterB , tc .read (producer , endpointID ), "B recreated the counter" )
389+
390+ // A releases. It must hit the orphaned counter, not B's live one.
391+ reqA .SchedulingResult = resA
392+ producer .ResponseBody (ctx , reqA , & tc .release , nil )
393+ require .Equal (t , tc .liveAfterReleaseA , tc .read (producer , endpointID ),
394+ "A's release must not discount B's live counter" )
395+
396+ // B releases. The live counter settles at 0 and never underflows.
397+ reqB .SchedulingResult = resB
398+ producer .ResponseBody (ctx , reqB , & tc .release , nil )
399+ require .Equal (t , int64 (0 ), tc .read (producer , endpointID ), "counter must settle at 0, never negative" )
400+ })
401+ }
373402}
374403
375404// TestInFlightLoadProducer_CrashWithHighLoadDoesNotUnderflow models a pod that crashes while
0 commit comments