feat(lmcache): implement decode first flow on lmcache connector when cache_hit_threshold field is present

[kyanokashi]

This PR updates the lmcache connector on the sidecar to decode first when cache_hit_threshold is present in the completion request.

The new flow goes as follows:

1. If cache_hit_threshold present; then decode first
2. If decode is successful write success response; return
3. If finish reason is cache_threshold that means the decode node didn't meet the cache hit threshold
4. Then we continue with the original flow of prefill, then decode

Test Report

Environment:

• Kind Cluster: llm-d-inference-scheduler-dev
• vLLM Simulator Image: kyanokashi/llm-d-inference-sim:dev
• Sidecar Connector: lmcache
• Log Level: --zap-log-level=4

---

Test Scenarios

Scenario 1: Non-Streaming WITHOUT cache_hit_threshold

Description: Basic request without lmcache protocol activation.

Result: ✅ PASS - Request proxied directly to decoder.

---

Scenario 2: Non-Streaming WITH cache_hit_threshold

Description: Request with cache_hit_threshold field, no cache_threshold triggered.

Result: ✅ PASS - LMCache protocol runs, response returned successfully.

---

Scenario 3: Non-Streaming WITH cache_hit_threshold AND X-Cache-Threshold: true

Description: Request configured to trigger cache_threshold finish_reason.

Result: ✅ PASS - LMCache protocol detects cache_threshold, triggers prefill→decode.

---

Scenario 4: Streaming WITHOUT cache_hit_threshold

Description: Streaming request without lmcache protocol activation.

Result: ✅ PASS - SSE chunks streamed directly from decoder.

---

Scenario 5: Streaming WITH cache_hit_threshold

Description: Streaming request with cache_hit_threshold, no header.

Result: ✅ PASS - LMCache protocol parses SSE chunks, streams through when no cache_threshold found.

---

Scenario 6: Streaming WITH cache_hit_threshold AND X-Cache-Threshold: true

Description: Streaming request with full lmcache configuration.

Result: ✅ PASS - LMCache protocol detects cache_threshold in SSE, triggers prefill→decode.

---

Failure Injection Test

To enable failure injection on the simulator:

--failure-injection-rate=100
--failure-types=rate_limit

Scenario 7: Streaming Decode Error (tryDecodeStreaming)

Request:

curl -s http://localhost:30080/v1/chat/completions \
  -H "Content-Type: application/json" \
  -d '{
    "model": "food-review",
    "messages": [{"role": "user", "content": "Hello"}],
    "max_tokens": 5,
    "cache_hit_threshold": 0.5,
    "stream": true
  }'

Path: tryDecodeStreaming → error

Result: ✅ PASS

• HTTP Status: 429
• Error returned as JSON (not SSE)

Response:

{
  "error": {
    "code": 429,
    "message": "Rate limit reached for food-review...",
    "type": "RateLimitError"
  }
}

---

Summary

Scenario	Streaming	cache_hit_threshold	X-Cache-Threshold	Result
1	No	No	No	✅ PASS
2	No	Yes	No	✅ PASS
3	No	Yes	Yes	✅ PASS
4	Yes	No	No	✅ PASS
5	Yes	Yes	No	✅ PASS
6	Yes	Yes	Yes	✅ PASS
7	Yes	Yes	No (error)	✅ PASS

All scenarios passed successfully.

[github-actions]

🚨 Unsigned commits detected! Please sign your commits.

For instructions on how to set up GPG/SSH signing and verify your commits,
please see GitHub Documentation.

[kfirwolfson]

Nice implementation, @kyanokashi. Please see comments inline, there are some things that I think should be modified for it to work properly.

[kfirwolfson]

This whole "if" code block and its contents should be removed. We don't want to start with "prefill". We want to start with "tryDecode" flow.

[kyanokashi]

This is to account for the temporary workaround @shmuelk mentioned he put in place for something related to istio. In that case we would only prefill, therefore, we don't want to call tryDecode.

[kfirwolfson]

I believe that if DP is in use, it's only for the Decode phase, and prefill works normally. @shmuelk please correct me if I am mistaken. Assuming this code will be removed soon, maybe we can avoid having a special handling for s.forwardDataParallel == true

[shmuelk]

This code is for the workaround we have for Istio 1.28.0.

Istio 1.28.1 has a fix for this issue. We will be removing this code in the future.

[kfirwolfson]

Should it be when sending to prefill or decode, @shmuelk ?

[kyanokashi]

So the istio workaround specifically intends to avoid decoding and only prefill for some scenarios.

I guess the question is whether the condition caused from the istio workaround and the cache_hit_threshold being present in the request could exist. Technically it shouldn't because that field is only relevant for decoding correct me if I'm wrong @kfirwolfson

[kfirwolfson]

I am not sure anyone will try running the connector_lmcache.py code with s.forwardDataParallel==True, let alone with cache_hit_threshold>0, before the s.forwardDataParallel code will be removed, so I don't think it matters much.
But for now, let's just remove this whole "if" statement. It's inaccurate for when s.forwardDataParallel==True and irrelevant for when s.forwardDataParallel==False.

[kyanokashi]

4c8659e

[kfirwolfson]

I suggest focusing on d-first use-case in this PR, and handing the Decode preemption problem (explained in detail in vLLM issue 24256 ), in separate PR.

[kfirwolfson]

Not sure how this would work with Streaming (vLLM online inference with partial responses). Would it?

[kyanokashi]

Hmmm this is tricky.

Previously we were passing a pass through writer to the decode proxy which was responsible for writing responses back to the client.

Now, because we need to parse the response to determine the finish reason, we use a buffered writer so we could read the first choice.

Let me explore some options. I'm thinking of either updating bufferedResponseWriter to support streaming or implementing a new writer type that handles the cache_threshold case specifically

[kyanokashi]

Here's what I ended up doing 88739c6

It's a bit complex, but couldn't think of a better way to do it.

Still need to test this

@kfirwolfson

[kfirwolfson]

nice work.

[kfirwolfson]

suggest removing the "(now deprecated)" text. We're working to enable it

[kyanokashi]

I'll remove it in the follow up refactor PR. It has massive refactoring and I prefer not changing the base if I can.

[kfirwolfson]

Overall, both logic and code look good to me, @elevran.

[kyanokashi]

@kfirwolfson @elevran @vMaroon what's left before we can merge this?

[kfirwolfson]

@kfirwolfson @elevran @vMaroon what's left before we can merge this?

Looks great to me.

[kyanokashi]

I went ahead an added coverage in the e2e tests d71b10b

[elevran]

@kyanokashi please fix the lint errors and ping me to approval. Other than that, I think this is ready to go in.
@kfirwolfson you may need to dismiss the Request Changes using the github UI before this can merge.

[kyanokashi]

@elevran I think you were referring to the test errors? Anyways they are fixed now

[kfirwolfson]

@elevran done. Approved from my pov.

[elevran]

@elevran I think you were referring to the test errors? Anyways they are fixed now

No. Was referring to the lint errors in the e2e test code introduced by this PR. Check the lint-and-test action :

Error: test/e2e/e2e_test.go:620:23: Error return value of `resp.Body.Close` is not checked (errcheck)
  	defer resp.Body.Close()
  	                     ^
  Error: test/e2e/e2e_test.go:647:23: Error return value of `resp.Body.Close` is not checked (errcheck)
  	defer resp.Body.Close()
  	                     ^
  Error: test/e2e/e2e_test.go:681:23: Error return value of `resp.Body.Close` is not checked (errcheck)
  	defer resp.Body.Close()
  	                     ^
  Error: test/e2e/e2e_test.go:715:23: Error return value of `resp.Body.Close` is not checked (errcheck)
  	defer resp.Body.Close()
  	                     ^
  Error: test/e2e/e2e_test.go:771:12: string-format: fmt.Sprintf can be replaced with string concatenation (perfsprint)
  	ginkgo.By(fmt.Sprintf("Getting request count from prefill pod: %s", prefillPodName))

[kyanokashi]

@elevran I think you were referring to the test errors? Anyways they are fixed now

No. Was referring to the lint errors in the e2e test code introduced by this PR. Check the lint-and-test action :

Error: test/e2e/e2e_test.go:620:23: Error return value of `resp.Body.Close` is not checked (errcheck)
  	defer resp.Body.Close()
  	                     ^
  Error: test/e2e/e2e_test.go:647:23: Error return value of `resp.Body.Close` is not checked (errcheck)
  	defer resp.Body.Close()
  	                     ^
  Error: test/e2e/e2e_test.go:681:23: Error return value of `resp.Body.Close` is not checked (errcheck)
  	defer resp.Body.Close()
  	                     ^
  Error: test/e2e/e2e_test.go:715:23: Error return value of `resp.Body.Close` is not checked (errcheck)
  	defer resp.Body.Close()
  	                     ^
  Error: test/e2e/e2e_test.go:771:12: string-format: fmt.Sprintf can be replaced with string concatenation (perfsprint)
  	ginkgo.By(fmt.Sprintf("Getting request count from prefill pod: %s", prefillPodName))

Ok, fixed now. There's an issue with the CI runs, where after they fail and I push a new commit, they will show as pending and I don't have access to the previous failed runs.

I tried running make lint locally which didn't show any errors because the linter was configured to not run on new code for some reason. I went ahead and made that configurable as well.

@elevran

[elevran]

/lgtm
/approve