The Progress Channel

The Progress Channel: broadcasting plan progress to N subscribers is easy until one subscriber is slow - then your "real-time" layer quietly becomes a memory leak or a brake on the plan itself. AnyCable years distilled to one rule: every channel names its BACKPRESSURE POLICY. This one offers two - :latest_wins for dashboards (drop stale frames), :every_event for auditors (bounded buffer, disconnect on overflow) - and proves each under a slow subscriber.

source on github

a real captured run

THE PROGRESS CHANNEL (backpressure is a policy, and policies have names)

  plan: 10 sequential tasks in 576ms (publish never blocked it)

  subscriber  policy        outcome after a slow session
  dashboard   latest_wins   holding 1 frame (the LATEST); 19 stale frames dropped
  auditor     every_event   DISCONNECTED at buffer 8 - it fell behind and gaps were unacceptable
  firehose    every_event   alive and current (it kept draining)

  the two policies are two PROMISES, and mixing them up is how
  real-time layers hurt people: a dashboard promised every-event
  buffers unboundedly behind one laggy browser tab until the
  publisher OOMs; an auditor promised latest-wins silently has
  holes exactly where the incident was. so the subscriber DECLARES
  which lie it can live with - stale (dashboard) or absent
  (auditor, disconnected LOUDLY so someone reconnects it) - and
  the publisher never blocks either way, because the plan's fibers
  have real work to do and instrumentation must never be the
  brake. name your backpressure policy or it names itself in
  production, and its name will be 'incident'.

source

# frozen_string_literal: true

# The Progress Channel: broadcasting plan progress to N subscribers
# is easy until one subscriber is slow - then your "real-time" layer
# quietly becomes a memory leak or a brake on the plan itself.
# AnyCable years distilled to one rule: every channel names its
# BACKPRESSURE POLICY. This one offers two - :latest_wins for
# dashboards (drop stale frames), :every_event for auditors (bounded
# buffer, disconnect on overflow) - and proves each under a slow
# subscriber.
#
#   bundle exec ruby examples/progress_channel.rb
#
# Runs offline; the slow subscriber is deliberately awful.

require class="s">"bundler/setup"
require class="s">"agentic"

Agentic.logger.level = class="y">:fatal

class ProgressChannel
  Subscriber = Struct.new(class="y">:name, class="y">:policy, class="y">:queue, class="y">:dropped, class="y">:dead, keyword_init: true)
  BUFFER_LIMIT = 8

  def initialize
    @subscribers = []
    @lock = Mutex.new
  end

  def subscribe(name, policy:)
    subscriber = Subscriber.new(name: name, policy: policy, queue: [], dropped: 0, dead: false)
    @lock.synchronize { @subscribers << subscriber }
    subscriber
  end

  # Publish never blocks and never fails the publisher - the plan's
  # fiber is doing real work; the channel absorbs or sheds, by policy
  def publish(event)
    @lock.synchronize do
      @subscribers.each do |sub|
        next if sub.dead

        case sub.policy
        when class="y">:latest_wins
          sub.dropped += sub.queue.size
          sub.queue.clear
          sub.queue << event
        when class="y">:every_event
          if sub.queue.size >= BUFFER_LIMIT
            sub.dead = true # an auditor with gaps is worse than no auditor
            sub.queue.clear
          else
            sub.queue << event
          end
        end
      end
    end
  end

  def hooks
    {
      task_slot_acquired: ->(task_id:, task:, waited:) { publish({at: class="y">:start, task: task.description}) },
      after_task_success: ->(task_id:, task:, result:, duration:) { publish({at: class="y">:done, task: task.description}) }
    }
  end
end

channel = ProgressChannel.new
dashboard = channel.subscribe(class="s">"dashboard", policy: class="y">:latest_wins)   # slow: renders at its own pace
auditor = channel.subscribe(class="s">"auditor", policy: class="y">:every_event)       # must see everything or nothing
firehose = channel.subscribe(class="s">"firehose", policy: class="y">:every_event)     # fast consumer, drains promptly

orchestrator = Agentic:class="y">:PlanOrchestrator.new(concurrency_limit: 2, lifecycle_hooks: channel.hooks)
previous = nil
10.times do |i|
  task = Agentic:class="y">:Task.new(description: class="s">"step-#{i}", agent_spec: {class="s">"name" => class="s">"s", class="s">"instructions" => class="s">"w"})
  orchestrator.add_task(task, previous ? [previous] : [], agent: ->(_t) {
    # the fast consumer drains during the plan; the others just... don't
    firehose.queue.clear
    sleep(0.055)
    class="y">:ok
  })
  previous = task
end

started = Process.clock_gettime(Process:class="y">:CLOCK_MONOTONIC)
orchestrator.execute_plan
elapsed = Process.clock_gettime(Process:class="y">:CLOCK_MONOTONIC) - started

puts class="s">"THE PROGRESS CHANNEL (backpressure is a policy, and policies have names)"
puts
puts format(class="s">"  plan: 10 sequential tasks in %dms (publish never blocked it)", (elapsed * 1000).round)
puts
puts format(class="s">"  %-11s %-13s %s", class="s">"subscriber", class="s">"policy", class="s">"outcome after a slow session")
puts format(class="s">"  %-11s %-13s holding %d frame (the LATEST); %d stale frames dropped",
  dashboard.name, dashboard.policy, dashboard.queue.size, dashboard.dropped)
puts format(class="s">"  %-11s %-13s DISCONNECTED at buffer %d - it fell behind and gaps were unacceptable",
  auditor.name, auditor.policy, ProgressChannel:class="y">:BUFFER_LIMIT)
puts format(class="s">"  %-11s %-13s alive and current (it kept draining)", firehose.name, firehose.policy)
puts
puts class="s">"  the two policies are two PROMISES, and mixing them up is how"
puts class="s">"  real-time layers hurt people: a dashboard promised every-event"
puts class="s">"  buffers unboundedly behind one laggy browser tab until the"
puts class="s">"  publisher OOMs; an auditor promised latest-wins silently has"
puts class="s">"  holes exactly where the incident was. so the subscriber DECLARES"
puts class="s">"  which lie it can live with - stale (dashboard) or absent"
puts class="s">"  (auditor, disconnected LOUDLY so someone reconnects it) - and"
puts class="s">"  the publisher never blocks either way, because the plan's fibers"
puts class="s">"  have real work to do and instrumentation must never be the"
puts class="s">"  brake. name your backpressure policy or it names itself in"
puts class="s">"  production, and its name will be 'incident'."