Status update publisher dies on oversize message

[natefoo]

Exception in thread consume-setup-amqp://main_pulsar:********@mq.galaxyproject.org:5671//main_pulsar?ssl=1:
}{"message": "ok"}{"message": "ok"}{"message": "ok"}{"message": "ok"}{"message": "ok"}{"message": "ok"}{"message": "ok"}{"message": "ok"}{"message": "ok"}{"message": "ok"}{"message": "ok"}{"message": "ok"}{"message": "ok"}{"message": "ok"}{"message": "ok"}{"message": "ok"}>
  File "/usr/lib64/python3.11/threading.py", line 1045, in _bootstrap_inner
    self.run()
  File "/srv/pulsar/main/venv/lib64/python3.11/site-packages/sentry_sdk/integrations/threading.py", line 99, in run
    return _run_old_run_func()
           ^^^^^^^^^^^^^^^^^^^
  File "/srv/pulsar/main/venv/lib64/python3.11/site-packages/sentry_sdk/integrations/threading.py", line 94, in _run_old_run_func
    reraise(*_capture_exception())
  File "/srv/pulsar/main/venv/lib64/python3.11/site-packages/sentry_sdk/utils.py", line 1640, in reraise
    raise value
  File "/srv/pulsar/main/venv/lib64/python3.11/site-packages/sentry_sdk/integrations/threading.py", line 92, in _run_old_run_func
    return old_run_func(self, *a, **kw)
           ^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/usr/lib64/python3.11/threading.py", line 982, in run
    self._target(*self._args, **self._kwargs)
  File "/srv/pulsar/main/venv/lib64/python3.11/site-packages/pulsar/messaging/bind_amqp.py", line 53, in drain
    __drain(name, queue_state, pulsar_exchange, callback)
  File "/srv/pulsar/main/venv/lib64/python3.11/site-packages/pulsar/messaging/bind_amqp.py", line 101, in __drain
    pulsar_exchange.consume(name, callback=callback, check=queue_state)
  File "/srv/pulsar/main/venv/lib64/python3.11/site-packages/pulsar/client/amqp_exchange.py", line 146, in consume
    connection.drain_events(timeout=self.__timeout)
  File "/srv/pulsar/main/venv/lib64/python3.11/site-packages/kombu/connection.py", line 341, in drain_events
    return self.transport.drain_events(self.connection, **kwargs)
           ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/srv/pulsar/main/venv/lib64/python3.11/site-packages/kombu/transport/pyamqp.py", line 171, in drain_events
    return connection.drain_events(**kwargs)
           ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/srv/pulsar/main/venv/lib64/python3.11/site-packages/amqp/connection.py", line 526, in drain_events
    while not self.blocking_read(timeout):
              ^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/srv/pulsar/main/venv/lib64/python3.11/site-packages/amqp/connection.py", line 532, in blocking_read
    return self.on_inbound_frame(frame)
           ^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/srv/pulsar/main/venv/lib64/python3.11/site-packages/amqp/method_framing.py", line 77, in on_frame
    callback(channel, msg.frame_method, msg.frame_args, msg)
  File "/srv/pulsar/main/venv/lib64/python3.11/site-packages/amqp/connection.py", line 538, in on_inbound_method
    return self.channels[channel_id].dispatch_method(
           ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/srv/pulsar/main/venv/lib64/python3.11/site-packages/amqp/abstract_channel.py", line 156, in dispatch_method
    listener(*args)
  File "/srv/pulsar/main/venv/lib64/python3.11/site-packages/amqp/channel.py", line 1629, in _on_basic_deliver
    fun(msg)
  File "/srv/pulsar/main/venv/lib64/python3.11/site-packages/kombu/messaging.py", line 656, in _receive_callback
    return on_m(message) if on_m else self.receive(decoded, message)
                                      ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/srv/pulsar/main/venv/lib64/python3.11/site-packages/kombu/messaging.py", line 622, in receive
    [callback(body, message) for callback in callbacks]
  File "/srv/pulsar/main/venv/lib64/python3.11/site-packages/kombu/messaging.py", line 622, in <listcomp>
    [callback(body, message) for callback in callbacks]
     ^^^^^^^^^^^^^^^^^^^^^^^
  File "/srv/pulsar/main/venv/lib64/python3.11/site-packages/pulsar/client/amqp_exchange.py", line 164, in __ack_callback
    self.publish(ack_queue, response)
  File "/srv/pulsar/main/venv/lib64/python3.11/site-packages/pulsar/client/amqp_exchange.py", line 236, in publish
    producer.publish(
  File "/srv/pulsar/main/venv/lib64/python3.11/site-packages/kombu/messaging.py", line 186, in publish
    return _publish(
           ^^^^^^^^^
  File "/srv/pulsar/main/venv/lib64/python3.11/site-packages/kombu/connection.py", line 556, in _ensured
    return fun(*args, **kwargs)
           ^^^^^^^^^^^^^^^^^^^^
  File "/srv/pulsar/main/venv/lib64/python3.11/site-packages/kombu/messaging.py", line 208, in _publish
    return channel.basic_publish(
           ^^^^^^^^^^^^^^^^^^^^^^
  File "/srv/pulsar/main/venv/lib64/python3.11/site-packages/amqp/channel.py", line 1791, in _basic_publish
    self.connection.drain_events(timeout=0)
  File "/srv/pulsar/main/venv/lib64/python3.11/site-packages/amqp/connection.py", line 526, in drain_events
    while not self.blocking_read(timeout):
              ^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/srv/pulsar/main/venv/lib64/python3.11/site-packages/amqp/connection.py", line 532, in blocking_read
    return self.on_inbound_frame(frame)
           ^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/srv/pulsar/main/venv/lib64/python3.11/site-packages/amqp/method_framing.py", line 53, in on_frame
    callback(channel, method_sig, buf, None)
  File "/srv/pulsar/main/venv/lib64/python3.11/site-packages/amqp/connection.py", line 538, in on_inbound_method
    return self.channels[channel_id].dispatch_method(
           ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/srv/pulsar/main/venv/lib64/python3.11/site-packages/amqp/abstract_channel.py", line 156, in dispatch_method
    listener(*args)
  File "/srv/pulsar/main/venv/lib64/python3.11/site-packages/amqp/channel.py", line 293, in _on_close
    raise error_for_code(
amqp.exceptions.PreconditionFailed: Basic.publish: (406) PRECONDITION_FAILED - message size 31486805 is larger than configured max size 16777216

[mvdbeek]

Do you know what's in the message ?

[mvdbeek]

(https://sentry.galaxyproject.org/share/issue/21c52d9004844f9eaa8f580d04f19e4d/)

[natefoo]

Correction, I believe this is a status update publish failure.

The size is due to a huge amount of binary data in stdout, produced by toolshed.g2.bx.psu.edu/repos/devteam/samtool_filter2/samtool_filter2/1.8+galaxy1. Viewing the tool_stdout, the stdout is a bam. The tool command is:

ln -s '/jetstream2/scratch/main/jobs/65367863/inputs/dataset_a.dat' input.bam && \
    ln -s '/jetstream2/scratch/main/jobs/65367863/inputs/metadata_metadata_b.dat' input.bai && \
    samtools view -U '/jetstream2/scratch/main/jobs/65367863/outputs/dataset_c.dat' -h   -b  -q 2 -L '/jetstream2/scratch/main/jobs/65367863/inputs/dataset_d.dat' input.bam

[natefoo]

So three issues:

1. Publisher dies and does not recover
2. Publish message is not shunted out of the queue so will be retried on restart
3. Whatever the tool is doing

Also hit #354 and this job is "lost" on restart since the active job file is removed.

[natefoo]

Well actually, this is confusing - the setup consumer is what's dying, but it's dying while publishing the ack, but the ack is a status update -> complete message.

[natefoo]

Yeah confirmed with py-spy this is the consume-setup-amqp://main_pulsar:********@mq.galaxyproject.org:5671//main_pulsar?ssl=1 thread, but the publish was for status_update:

{
  "job_id": "65367863",
  "complete": "true",
  "status": "complete",
  "returncode": 0,
  "stdout": "\u001f�...",
  "stderr": "",
  "job_stdout": "",
  "job_stderr": "",
  "working_directory": "/jetstream2/scratch/main/jobs/65367863/working",
  "metadata_directory": "/jetstream2/scratch/main/jobs/65367863/metadata",
  "job_directory": "/jetstream2/scratch/main/jobs/65367863",
  "working_directory_contents": [
    "input.bai",
    "input.bam"
  ],
  "metadata_directory_contents": [
    "job_stderr",
    "__instrument_core_galaxy_slots",
    "metadata_results_output1",
    "metadata_out_output1",
    "tool_stdout",
    "job_stdout",
    "__instrument_core_container",
    "params.json",
    "__instrument_meminfo_meminfo",
    "__instrument_uname_uname",
    "__instrument_cpuinfo_cpuinfo",
    "__instrument_cgroup__metrics",
    "tool_stderr",
    "metadata_temp_file_k5vugjsm",
    "set.py",
    "metadata_kwds_output1",
    "metadata_override_output1",
    "__instrument_core_epoch_end",
    "__instrument_core_epoch_start",
    "__instrument_core_galaxy_memory_mb",
    "memory_statement.log",
    "outputs_new/jobs_attrs.txt",
    "outputs_new/datasets_attrs.txt",
    "outputs_new/implicit_dataset_conversions.txt",
    "outputs_new/library_folders_attrs.txt",
    "outputs_new/collections_attrs.txt",
    "outputs_new/libraries_attrs.txt",
    "outputs_new/datasets_attrs.txt.provenance",
    "outputs_new/export_attrs.txt",
    "outputs_new/tool.xml",
    "outputs_new/invocation_attrs.txt"
  ],
  "outputs_directory_contents": [
    "dataset_c.dat"
  ],
  "job_directory_contents": [
    "tool_id",
    "tool_script.sh",
    "home",
    "tool_files",
    "return_code",
    "tmp",
    "tool_version",
    "tmp.20250228-101841",
    "final_status",
    "working",
    "preprocessed",
    "launch_config",
    "metadata",
    "running",
    "inputs",
    "postprocessed",
    "external_id",
    "galaxy_65367863.ec",
    "outputs",
    "command.sh",
    "configs"
  ],
  "system_properties": {
    "separator": "/",
    "galaxy_home": "/cvmfs/main.galaxyproject.org/galaxy",
    "galaxy_virtual_env": "/cvmfs/main.galaxyproject.org/venv",
    "galaxy_datatypes_config_file": "/cvmfs/main.galaxyproject.org/galaxy/config/datatypes_conf.xml.sample"
  },
  "pulsar_version": "0.15.7.dev0",
  "realized_dynamic_file_sources": [],
  "acknowledge_uuid": "a64076bc-f5bd-11ef-9933-fa163ef566a9",
  "acknowledge_queue": "status_update_ack",
  "acknowledge_submit_queue": "status_update"
}

[natefoo]

I don't know why but I'm getting this semi-regularly now. I think we used to recover more gracefully from this (and I don't understand why it's killing the setup consumer).

[mvdbeek]

Can you lower the max size to maybe 1MB instead of 8 ?

[natefoo]

Is there a config option for it?

[mvdbeek]

https://github.com/search?q=repo%3Agalaxyproject%2Fusegalaxy-playbook%20maximum_stream_size&type=code

[natefoo]

Facepalm