pq writer can become stuck after precisely filling queue max events

[yaauie]

Logstash information:

Please include the following information:

1. Logstash version (e.g. bin/logstash --version): 8.8.0..main
2. Logstash installation source (e.g. built from source, with a package manager: DEB/RPM, expanded from tar or zip archive, docker): any
3. How is Logstash being run (e.g. as a service/service manager: systemd, upstart, etc. Via command line, docker/kubernetes): any

Plugins installed: (bin/logstash-plugin list --verbose)

N/A

JVM (e.g. java -version):

Bundled, but N/A

OS version (uname -a if on a Unix-like system):

MacOS, but N/A

Description of the problem including expected versus actual behavior:

When a pipeline is configured with a queue.max_bytes that matches its queue.page_capacity,
and writes an event whose serialized form precisely fits in the remaining capacity of the queue,
the single writer can become blocked indefinitely.

After writing the serialized bytes to the head page, if there is not room for more events,
the writer will enter a loop in which it blocks until notified on the notFull condition variable, repeating the loop until such time as there is room for more bytes.

The notFull condition is only signaled when:

• A reader observed that the queue was over its max unread limit (event quantity) before beginning its read (e.g., NOT when the queue had been over its bytes capacity)
• A non-empty batch from a TAIL page is fully-ack'd
• The queue is closed

But signalling the notFull condition is not enough; the queue will remain "full" (insufficient capacity to write a single byte) until capacity is recovered by a tail page becoming fully-acked and removed.

In some conditions a second writer could arrive and effectively break the block by rolling the current head page over into a tail page before the events had been acknowledged (temporarily exceeding queue.max_bytes), but once that window is missed the block is permanent.

Steps to reproduce:

1. Configure Logstash with a small persisted queue to improve the speed at which we can create the failure condition:

   queue.type: persisted
   queue.max_bytes: 32768
   queue.page_capacity: 32768
   

2. Create a ruby script that generates randomized-length JSON payloads to its stdout, so that each time the queue nears capacity we have a randomized chance of serializing to the precicely-correct length:

   #!/usr/bin/env ruby
   require "json"
   loop do
     puts({"k" => ("v"*Random.rand(1..64))}.to_json)
   end
   

3. Pipe the output of the ruby script to Logstash configured without ECS compatibility (to avoid event.original)

   ./generate_random_json.rb | "${LOGSTASH_HOME}/bin/logstash" --debug -w1 -b1 --pipeline.ecs_compatibility=disabled -e 'input { stdin { codec => json_lines } } output { stdout { codec => dots } }'
   

4. In a separate tab, use watch to observe the PQ until its queue_size_in_bytes exactly matches its max_queue_size_in_bytes, observing that the events also stop flowing in a state where (a) the queue backpressure is complete and (b) the workers are functionally starved.

   watch --color '(curl --silent -XGET http://localhost:9600/_node/stats | jq --color-output ".pipelines.main | pick(.queue, .flow)")'
   

   {
     "queue": {
       "data": {
         "free_space_in_bytes": 749054627840,
         "path": "/Users/rye/src/elastic/sdh-logstash/1486-many-p2p-backpressure-propagation-one-host-but-not-others/20240520/repro/logstash-8.8.0/data/queue/main",
         "storage_type": "apfs"
       },
       "type": "persisted",
       "capacity": {
         "queue_size_in_bytes": 32768,
         "max_queue_size_in_bytes": 32768,
         "max_unread_events": 0,
         "page_capacity_in_bytes": 32768
       },
       "events": 0,
       "events_count": 0,
       "queue_size_in_bytes": 32768,
       "max_queue_size_in_bytes": 32768
     },
     "flow": {
       "output_throughput": {
         "current": 0.0,
         "last_1_minute": 0.0,
         "last_5_minutes": 0.0,
         "lifetime": 9.131
       },
       "queue_backpressure": {
         "current": 1.0,
         "last_1_minute": 1.0,
         "last_5_minutes": 1.0,
         "lifetime": 0.9994
       },
       "queue_persisted_growth_events": {
         "current": 0.0,
         "last_1_minute": 0.0,
         "last_5_minutes": 0.0,
         "lifetime": 0.0
       },
       "queue_persisted_growth_bytes": {
         "current": 0.0,
         "last_1_minute": 0.0,
         "last_5_minutes": 0.0,
         "lifetime": 45.23
       },
       "input_throughput": {
         "current": 0.0,
         "last_1_minute": 0.0,
         "last_5_minutes": 0.0,
         "lifetime": 9.131
       },
       "worker_concurrency": {
         "current": 0.0,
         "last_1_minute": 0.00006219,
         "last_5_minutes": 0.00004845,
         "lifetime": 0.0003117
       },
       "filter_throughput": {
         "current": 0.0,
         "last_1_minute": 0.0,
         "last_5_minutes": 0.0,
         "lifetime": 9.131
       }
     }
   }

Provide logs (if relevant):

Workarounds

I have thus far been unable to replicate the issue when queue.max_bytes`` is at least _double_ queue.page_capacity(whose default is 64MiB, or67108864`).

[yaauie]

We can exhibit this in a pipeline-to-pipeline with a simpler config:

 - pipeline.id: entrypoint
   pipeline.ecs_compatibility: disabled
   pipeline.batch.size: 1
   pipeline.workers: 1
   config.string: |
     input { generator { } }
     filter {
       ruby {
         code => "event.set('k', 10_000.times.map { 'a'*::Random.rand(1..64) })"
       }
       split {
         field => "k"
       }
     }
     output {
       pipeline { send_to => [one] }
     }
 - pipeline.id: downstream
   pipeline.ecs_compatibility: disabled
   queue.type: persisted
   queue.max_bytes: 32768
   queue.page_capacity: 32768
   config.string: |
     input { pipeline { address => one } }
     output { stdout { codec => dots } }

-- pipelines.yml

and:

"${LOGSTASH_HOME}/bin/logstash" --debug

The watch/curl/jq command becomes:

watch --color '(curl --silent -XGET http://localhost:9600/_node/stats | jq --color-output ".pipelines.downstream | pick(.queue, .flow)")'