Websocket: Using PerMessageDeflateExtension and flush in batchMode send FLUSH_FRAME to client.

[jehuty0shift]

Jetty version(s)
websocket-jakarta-server 11.0.15 (possibly 12.x)

Java version/vendor (use: java -version)
openjdk version "17.0.7" 2023-04-18 LTS
OpenJDK Runtime Environment Zulu17.42+21-CRaC-CA (build 17.0.7+7-LTS)
OpenJDK 64-Bit Server VM Zulu17.42+21-CRaC-CA (build 17.0.7+7-LTS, mixed mode, sharing)

OS type/version
Ubuntu 22.04.2 LTS

Description
Calling RemoteEndpoint.flushBatch() with batching enabled causes the FLUSH_FRAME to be propagated all the way to the client when "permessage-deflate" has been activated (and so PerMessageDeflateExtension is responsible for sending frames).

WebSocketCoreSession injects a FrameFlusher.FLUSH_FRAME in the pipeline to force flushing, however this frame when used with the deflate extension ends up being sent to the client due to a chunk copy when Deflating. This new propagated Frame then fails the equlity test in FrameFlusher, and the resulting Frame ends being sent on the Network.

Here is the full listing of logs in Debug when using permessage-deflate (I send a text Frame then I basically call flushBatch). Note the reference for the BINARY Framed changed from BINARY@375b9d69 to BINARY@56bbfc33 thus escaping FLUSH_FRAME detection

sendFrame(TEXT@3d4dc6b3[len=415,fin=true,rsv=000,m=null]HeapByteBuffer@65d23c28[p=0,l=415,c=415,r=415]={<<<{"message":"{\\"_new_era\\"...,\\"version\\":\\"1.1\\"}"}>>>}, FutureCallback@55a1c1e7{false,false}, true)
DEBUG 12:09:47,123 Queuing TEXT@3d4dc6b3[len=415,fin=true,rsv=000,m=null]HeapByteBuffer@65d23c28[p=0,l=415,c=415,r=415]={<<<{"message":"{\\"_new_era\\"...,\\"version\\":\\"1.1\\"}"}>>>}
DEBUG 12:09:47,123 onFrame TEXT@3d4dc6b3[len=415,fin=true,rsv=000,m=null]HeapByteBuffer@65d23c28[p=0,l=415,c=415,r=415]={<<<{"message":"{\\"_new_era\\"...,\\"version\\":\\"1.1\\"}"}>>>}
DEBUG 12:09:47,123 Extending out TEXT@3d4dc6b3[len=415,fin=true,rsv=000,m=null]HeapByteBuffer@65d23c28[p=0,l=415,c=415,r=415]={<<<{"message":"{\\"_new_era\\"...,\\"version\\":\\"1.1\\"}"}>>>} Flusher@5fd1d126[PROCESSING] true
DEBUG 12:09:47,123 Queuing TEXT@3d4dc6b3[len=415,fin=true,rsv=000,m=null]HeapByteBuffer@65d23c28[p=0,l=415,c=415,r=415]={<<<{"message":"{\\"_new_era\\"...,\\"version\\":\\"1.1\\"}"}>>>}
DEBUG 12:09:47,124 onFrame TEXT@3d4dc6b3[len=415,fin=true,rsv=000,m=null]HeapByteBuffer@65d23c28[p=0,l=415,c=415,r=415]={<<<{"message":"{\\"_new_era\\"...,\\"version\\":\\"1.1\\"}"}>>>}
DEBUG 12:09:47,124 transform TEXT@3d4dc6b3[len=415,fin=true,rsv=000,m=null]HeapByteBuffer@65d23c28[p=0,l=415,c=415,r=415]={<<<{"message":"{\\"_new_era\\"...,\\"version\\":\\"1.1\\"}"}>>>}
DEBUG 12:09:47,125 Compressed 267 bytes Deflater[finished=false,read=415,written=267,in=415,out=267]
DEBUG 12:09:47,125 Compressed 0 bytes Deflater[finished=false,read=415,written=267,in=415,out=267]
DEBUG 12:09:47,125 payload (TAIL_DROP_FIN_ONLY) = HeapByteBuffer@505baf4a[p=0,l=263,c=8192,r=263]={<<<T\x90\xDbj\xC30\x0c\x86_\xC5\xF8z\x98$\xCd\xB1\x97\x83\xBdE\xC0x\xAe\xD2\x98...\x8f};t]\xD3\x0e/\x13En\x10\xA2A\x7f\xDcv\xBf\xE9\x83?\xBe\x00>>>\x00\x00\xFf\xFf\x00\x00\x00\x00\x00...\x00\x00\x00\x00\x00\x00\x00}
DEBUG 12:09:47,125 Compressed TEXT@3d4dc6b3[len=415,fin=true,rsv=000,m=null]HeapByteBuffer@65d23c28[p=0,l=415,c=415,r=415]={<<<{"message":"{\\"_new_era\\"...,\\"version\\":\\"1.1\\"}"}>>>}: payload:263
DEBUG 12:09:47,125 nextOutgoingFrame(TEXT@5b774ec7[len=263,fin=true,rsv=100,m=null]HeapByteBuffer@505baf4a[p=0,l=263,c=8192,r=263]={<<<T\x90\xDbj\xC30\x0c\x86_\xC5\xF8z\x98$\xCd\xB1\x97\x83\xBdE\xC0x\xAe\xD2\x98...\x8f};t]\xD3\x0e/\x13En\x10\xA2A\x7f\xDcv\xBf\xE9\x83?\xBe\x00>>>\x00\x00\xFf\xFf\x00\x00\x00\x00\x00...\x00\x00\x00\x00\x00\x00\x00})
DEBUG 12:09:47,125 Enqueued Entry{TEXT@5b774ec7[len=263,fin=true,rsv=100,m=null]HeapByteBuffer@505baf4a[p=0,l=263,c=8192,r=263]={<<<T\x90\xDbj\xC30\x0c\x86_\xC5\xF8z\x98$\xCd\xB1\x97\x83\xBdE\xC0x\xAe\xD2\x98...\x8f};t]\xD3\x0e/\x13En\x10\xA2A\x7f\xDcv\xBf\xE9\x83?\xBe\x00>>>\x00\x00\xFf\xFf\x00\x00\x00\x00\x00...\x00\x00\x00\x00\x00\x00\x00},org.eclipse.jetty.util.Callback$6@46cf1e1a,true} to Flusher@3d2d4a78[IDLE][queueSize=1,aggregate=null]
DEBUG 12:09:47,125 Flushing Flusher@3d2d4a78[PROCESSING][queueSize=1,aggregate=null]
DEBUG 12:09:47,125 Flusher@3d2d4a78[PROCESSING][queueSize=0,aggregate=HeapByteBuffer@3a775492[p=0,l=267,c=4096,r=267]={<<<\xC1~\x01\x07T\x90\xDbj\xC30\x0c\x86_\xC5\xF8z\x98$\xCd\xB1\x97\x83\xBdE\xC0...\x8f};t]\xD3\x0e/\x13En\x10\xA2A\x7f\xDcv\xBf\xE9\x83?\xBe\x00>>>\x00\x00\x00\x00\x00\x00\x00\x00\x00...\x00\x00\x00\x00\x00\x00\x00}] processed 1 entries flush=false batch=HeapByteBuffer@3a775492[p=0,l=267,c=4096,r=267]={<<<\xC1~\x01\x07T\x90\xDbj\xC30\x0c\x86_\xC5\xF8z\x98$\xCd\xB1\x97\x83\xBdE\xC0...\x8f};t]\xD3\x0e/\x13En\x10\xA2A\x7f\xDcv\xBf\xE9\x83?\xBe\x00>>>\x00\x00\x00\x00\x00\x00\x00\x00\x00...\x00\x00\x00\x00\x00\x00\x00}: [Entry{TEXT@5b774ec7[len=263,fin=true,rsv=100,m=null]HeapByteBuffer@505baf4a[p=0,l=263,c=8192,r=263]={<<<T\x90\xDbj\xC30\x0c\x86_\xC5\xF8z\x98$\xCd\xB1\x97\x83\xBdE\xC0x\xAe\xD2\x98...\x8f};t]\xD3\x0e/\x13En\x10\xA2A\x7f\xDcv\xBf\xE9\x83?\xBe\x00>>>\x00\x00\xFf\xFf\x00\x00\x00\x00\x00...\x00\x00\x00\x00\x00\x00\x00},org.eclipse.jetty.util.Callback$6@46cf1e1a,true}]
DEBUG 12:09:47,125 Flushing Flusher@3d2d4a78[PROCESSING][queueSize=0,aggregate=HeapByteBuffer@3a775492[p=0,l=267,c=4096,r=267]={<<<\xC1~\x01\x07T\x90\xDbj\xC30\x0c\x86_\xC5\xF8z\x98$\xCd\xB1\x97\x83\xBdE\xC0...\x8f};t]\xD3\x0e/\x13En\x10\xA2A\x7f\xDcv\xBf\xE9\x83?\xBe\x00>>>\x00\x00\x00\x00\x00\x00\x00\x00\x00...\x00\x00\x00\x00\x00\x00\x00}]
DEBUG 12:09:47,125 Flusher@3d2d4a78[PROCESSING][queueSize=0,aggregate=HeapByteBuffer@3a775492[p=0,l=267,c=4096,r=267]={<<<\xC1~\x01\x07T\x90\xDbj\xC30\x0c\x86_\xC5\xF8z\x98$\xCd\xB1\x97\x83\xBdE\xC0...\x8f};t]\xD3\x0e/\x13En\x10\xA2A\x7f\xDcv\xBf\xE9\x83?\xBe\x00>>>\x00\x00\x00\x00\x00\x00\x00\x00\x00...\x00\x00\x00\x00\x00\x00\x00}] processed 0 entries flush=false batch=HeapByteBuffer@3a775492[p=0,l=267,c=4096,r=267]={<<<\xC1~\x01\x07T\x90\xDbj\xC30\x0c\x86_\xC5\xF8z\x98$\xCd\xB1\x97\x83\xBdE\xC0...\x8f};t]\xD3\x0e/\x13En\x10\xA2A\x7f\xDcv\xBf\xE9\x83?\xBe\x00>>>\x00\x00\x00\x00\x00\x00\x00\x00\x00...\x00\x00\x00\x00\x00\x00\x00}: []
DEBUG 12:09:47,125 notifyCallbackSuccess Flusher@5fd1d126[PROCESSING]
DEBUG 12:09:47,125 notifyCallbackSuccess FutureCallback@55a1c1e7{false,false}
DEBUG 12:09:47,626 Queuing BINARY@375b9d69[len=0,fin=true,rsv=000,m=null]
DEBUG 12:09:47,626 onFrame BINARY@375b9d69[len=0,fin=true,rsv=000,m=null]
DEBUG 12:09:47,626 Extending out BINARY@375b9d69[len=0,fin=true,rsv=000,m=null] Flusher@5fd1d126[PROCESSING] false
DEBUG 12:09:47,626 Queuing BINARY@375b9d69[len=0,fin=true,rsv=000,m=null]
DEBUG 12:09:47,627 onFrame BINARY@375b9d69[len=0,fin=true,rsv=000,m=null]
DEBUG 12:09:47,627 transform BINARY@375b9d69[len=0,fin=true,rsv=000,m=null]
DEBUG 12:09:47,627 Compressed 0 bytes Deflater[finished=false,read=415,written=267,in=415,out=267]
DEBUG 12:09:47,627 Compressed BINARY@375b9d69[len=0,fin=true,rsv=000,m=null]: payload:1
DEBUG 12:09:47,627 nextOutgoingFrame(BINARY@56bbfc33[len=1,fin=true,rsv=100,m=null]HeapByteBuffer@68ba0730[p=0,l=1,c=1,r=1]={<<<\x00>>>})
DEBUG 12:09:47,627 Enqueued Entry{BINARY@56bbfc33[len=1,fin=true,rsv=100,m=null]HeapByteBuffer@68ba0730[p=0,l=1,c=1,r=1]={<<<\x00>>>},org.eclipse.jetty.util.Callback$6@68691d71,false} to Flusher@3d2d4a78[IDLE][queueSize=1,aggregate=HeapByteBuffer@3a775492[p=0,l=267,c=4096,r=267]={<<<\xC1~\x01\x07T\x90\xDbj\xC30\x0c\x86_\xC5\xF8z\x98$\xCd\xB1\x97\x83\xBdE\xC0...\x8f};t]\xD3\x0e/\x13En\x10\xA2A\x7f\xDcv\xBf\xE9\x83?\xBe\x00>>>\x00\x00\x00\x00\x00\x00\x00\x00\x00...\x00\x00\x00\x00\x00\x00\x00}]
DEBUG 12:09:47,627 Flushing Flusher@3d2d4a78[PROCESSING][queueSize=1,aggregate=HeapByteBuffer@3a775492[p=0,l=267,c=4096,r=267]={<<<\xC1~\x01\x07T\x90\xDbj\xC30\x0c\x86_\xC5\xF8z\x98$\xCd\xB1\x97\x83\xBdE\xC0...\x8f};t]\xD3\x0e/\x13En\x10\xA2A\x7f\xDcv\xBf\xE9\x83?\xBe\x00>>>\x00\x00\x00\x00\x00\x00\x00\x00\x00...\x00\x00\x00\x00\x00\x00\x00}]
DEBUG 12:09:47,627 Flusher@3d2d4a78[PROCESSING][queueSize=0,aggregate=HeapByteBuffer@3a775492[p=0,l=269,c=4096,r=269]={<<<\xC1~\x01\x07T\x90\xDbj\xC30\x0c\x86_\xC5\xF8z\x98$\xCd\xB1\x97\x83\xBdE\xC0...;t]\xD3\x0e/\x13En\x10\xA2A\x7f\xDcv\xBf\xE9\x83?\xBe\x00\xC2\x01>>>\x00\x00\x00\x00\x00\x00\x00\x00\x00...\x00\x00\x00\x00\x00\x00\x00}] processed 1 entries flush=true batch=HeapByteBuffer@3a775492[p=0,l=269,c=4096,r=269]={<<<\xC1~\x01\x07T\x90\xDbj\xC30\x0c\x86_\xC5\xF8z\x98$\xCd\xB1\x97\x83\xBdE\xC0...;t]\xD3\x0e/\x13En\x10\xA2A\x7f\xDcv\xBf\xE9\x83?\xBe\x00\xC2\x01>>>\x00\x00\x00\x00\x00\x00\x00\x00\x00...\x00\x00\x00\x00\x00\x00\x00}: [Entry{BINARY@56bbfc33[len=1,fin=true,rsv=100,m=null]HeapByteBuffer@68ba0730[p=0,l=1,c=1,r=1]={<<<\x00>>>},org.eclipse.jetty.util.Callback$6@68691d71,false}]
DEBUG 12:09:47,627 Flushing Flusher@3d2d4a78[PROCESSING][queueSize=0,aggregate=HeapByteBuffer@3a775492[p=269,l=269,c=4096,r=0]={\xC1~\x01\x07T\x90\xDbj\xC3...\xE9\x83?\xBe\x00\xC2\x01<<<>>>\x00\x00\x00\x00\x00\x00\x00\x00\x00...\x00\x00\x00\x00\x00\x00\x00}]
DEBUG 12:09:47,627 Flusher@3d2d4a78[PROCESSING][queueSize=0,aggregate=HeapByteBuffer@3a775492[p=0,l=0,c=4096,r=0]={<<<>>>\xC1~\x01\x07T\x90\xDbj\xC3...\x00\x00\x00\x00\x00\x00\x00}] processed 0 entries flush=false batch=HeapByteBuffer@3a775492[p=0,l=0,c=4096,r=0]={<<<>>>\xC1~\x01\x07T\x90\xDbj\xC3...\x00\x00\x00\x00\x00\x00\x00}: []
DEBUG 12:09:47,627 notifyCallbackSuccess Flusher@5fd1d126[PROCESSING]
DEBUG 12:09:47,628 notifyCallbackSuccess FutureCallback@3b275c0{false,false}

I am not confident enough on the Jetty architecture to submit a proper patch since there is multiple fix possible for that (from "Create a proper FlushFrame object extending Frame to use 'equals' instead of '=='" to "having PerMessageDeflate to detect Flush frame"). But i'm willingful to help fix the problem and PR.

How to reproduce?

• Start a websocket server
• Start a websocket client with permessage-deflate extension activated
• Send a Message from server to client
• Use flushBatch()
• Observe 2 messages (TEXT and BINARY) on Client side.