Decode each checkpoint blob once per list page

Eduardo Bonetrequested to merge
631232-decode-each-checkpoint-blob-once-per-list-pag into master

What does this MR do and why?

The Duo Workflow checkpoint list endpoint now decodes each checkpoint blob once per page. With accept_compressed=true, it also keeps only one rebuilt checkpoint in memory at a time. Before, checkpoints in the same group each decoded the same large snapshot blobs, so a page of 20 decoded them up to 20 times and held all 20 rebuilt checkpoints until the response was ready.

On a synthetic workflow with 60 checkpoints, a compressed per_page=20 request allocates 194 MB instead of 436 MB. Memory still held before rendering drops from 140 MB to 45 MB. Response bodies are byte-identical to master.

Benchmark

Local, synthetic workflow: 60 checkpoints in one current_thread group, 30 KB and 15 KB per-step deltas, about 1.6 MB of compressed blobs. Measured through the real reconstruct_page! helper plus compression, with MemoryProfiler for allocations and ObjectSpace.memsize_of_all after a full GC for live memory. "Allocated" is the same kind of metric as mem_total_bytes in the Rails API logs. "Live before render" is memory still reachable after reconstruct and compress, before the response is serialized. No peak-memory number was measured. These are local synthetic numbers, not production.

Page Allocated, master Allocated, this MR Live before render, master Live before render, this MR
per_page=1 38 MB 34 MB 10 MB 6 MB
per_page=5 133 MB 64 MB 44 MB 18 MB
per_page=20 436 MB 194 MB 140 MB 45 MB
Design notes

ChannelValuesReconstructor.new(blobs, decoded: {}) takes an optional Hash that caches each blob's decoded value. reconstruct_page! passes one Hash to every reconstructor on the page. The key is the blob record: ActiveRecord equality uses the primary key, so a row loaded twice (the anchor-miss refetch) still hits the cache.

Only #channel_values reads the cache. It's the only path the list endpoint uses, and #channel_history stamps entries in place, so it must not share decoded values.

The fold grows the running value in place (Array#concat, or per-key concat for a dict-of-list channel). A shared decoded value must not grow, so the fold now works on a copy of each decoded value: Array#dup, or for a Hash a new Hash with each Array value dup'd. The copy is shallow, so message entries are shared, and the fold never mutates entries. Copying every decoded value, not only the anchor, also covers a channel that starts as nil: its first delta becomes the running value.

With accept_compressed=true, reconstruct_page! compresses each checkpoint right after rebuilding it and stores only compressed_checkpoint, so header.checkpoint keeps its slim value. The entity exposes checkpoint only when compressed_checkpoint is blank, so the response is unchanged. Without accept_compressed, the entity needs the full checkpoint, so headers keep it as before. The legacy (non-reconstruct) path is unchanged.

The issue asks to ship this after the sibling issues below and to re-measure first. Those may remove most of the cost for the main caller (the AI Gateway stop-recovery walk), but this change helps every caller of the endpoint.

References

Screenshots or screen recordings

No UI change.

How to set up and validate locally

The endpoint is internal (used by the Duo Workflow Service) and has no UI path.

  1. Run the updated specs: bundle exec rspec ee/spec/lib/gitlab/duo_workflow/channel_values_reconstructor_spec.rb ee/spec/requests/api/ai/duo_workflows/workflows_internal_spec.rb

  2. In a GDK rails console, pick a workflow with checkpoint headers and blobs, and print a digest of its compressed first page:

    mod = API::Ai::DuoWorkflows::WorkflowsInternal.namespace_stackable(:helpers).find { |m| m.method_defined?(:reconstruct_page!) }
helper = Object.new.extend(mod)
workflow = Ai::DuoWorkflows::Workflow.find(WORKFLOW_ID)
headers = workflow.checkpoint_headers.in_reverse_checkpoint_order.limit(20).to_a
helper.reconstruct_page!(workflow, headers, compress: true)
Digest::SHA256.hexdigest(API::Entities::Ai::DuoWorkflows::Checkpoint.represent(headers).to_json)

  3. Repeat on master. There, replace the reconstruct_page! line with:

    helper.reconstruct_page!(workflow, headers)
headers.each { |h| h.compressed_checkpoint = helper.compress_checkpoint(h.checkpoint) }

    Expect: the same digest on both branches.

Benchmark reproduction

The numbers above came from a local script that wraps step 2 in MemoryProfiler.report for allocations and samples ObjectSpace.memsize_of_all after a full GC for live memory. The synthetic workflow was seeded through Ai::DuoWorkflows::CreateCheckpointService with one group-start snapshot per channel and 59 checkpoints of conversation_history and ui_chat_log deltas. Both scripts are untracked; I can share them on request.

MR acceptance checklist

Evaluate this MR against the MR acceptance checklist. It helps you analyze changes to reduce risks in quality, performance, reliability, security, and maintainability.

Merge request reports