Async actions

The endpoint runs your action inside a transaction and renders the reply there, while the queue adapter publishes on commit. So any reply.morph after an enqueue renders from a database the job has not touched yet:

def restore_all
  count = BatchRestoreService.call(bulk_payment: @bulk_payment)  # fans out N jobs
  reply.morph.flash(:notice, "Putting #{count} back…")           # renders the PRE-job world
end

The morph re-renders every row with a live "Put back" button for work already requested, sitting directly beside the "Queued 177" flash the same reply emitted.

class ReconcileQueue < ApplicationComponent
  include Phlex::Reactive::Component

  reactive_collection :unreconcilable,
    item: TransferRow, container: "unreconcilable",
    count: "unreconcilable-count", empty: NothingToReconcile,
    size: -> { @bulk_payment.transfers.unreconcilable.count }

  action :re_execute, params: { transfer_id: :integer }

  def re_execute(transfer_id:)
    transfer = @bulk_payment.transfers.re_executable.find(transfer_id)
    authorize! transfer, :update?
    reply.pending(transfer, in: :unreconcilable, job: ReExecuteJob, args: [transfer.id])
  end
end

class ReExecuteJob < ApplicationJob
  include Phlex::Reactive::Settles

  def perform(transfer_id)            # signature UNCHANGED
    transfer = Transfer.find(transfer_id)
    result   = Transfers::ReExecuteService.call(transfer:)

    reactive_settle do |s|
      if result.success?
        s.remove(transfer)                # row + count + empty-state
      else
        s.replace(transfer)               # back to actionable
        s.flash(:alert, result.error)     # the operator learns the outcome
      end
    end
  end
end

And the case that motivated the whole feature — work that moves a record between two lists — is one call:

reactive_settle { |s| s.move(transfer, from: :declined, to: :unreconcilable) }

It deliberately does not re-render the container — that is the bug. It emits three things:

1. a data-reactive-pending="true" + aria-busy="true" marker on every target, over the existing reactive:js op lane;
2. one subscription directive, anchored on the container, opening a single durable one-shot stream that all N settles share;
3. an inert reactive:token refresh, so the container's signed token rolls forward and the list is not act-once-only.

One CSS rule covers the whole pending vocabulary:

[data-reactive-pending] { opacity: .5; pointer-events: none; }

The block form is what makes the motivating case work — the enqueue usually lives in a service object, not the action:

def restore_all
  restorable = @bulk_payment.transfers.restorable.to_a
  reply.pending(restorable, in: :declined, peers: true) do
    BatchRestoreService.call(bulk_payment: @bulk_payment)
  end.flash(:notice, "Putting #{restorable.size} back…")
end

The markers dim a live row. When the pending state should be real markup — a "Queued" badge, a button that is genuinely gone — define pending_template on the row component (private is fine). reply.pending then also replaces each such row with it, just before marking it. A row without the hook is byte-identical to the marker-only reply.

class TransferRow < ApplicationComponent
  include Phlex::Reactive::Streamable
  include Phlex::Reactive::Component

  def view_template
    li(**mix(reactive_attrs, id:)) { span { @transfer.label }; button(**on(:retry)) { "Retry" } }
  end

  private

  def pending_template
    li(**mix(reactive_attrs, id:)) { span { @transfer.label }; span(class: "badge") { "Queued" } }
  end
end

• Keep the root's id — the settle targets it. A pending_template without it raises before anything is enqueued.
• Keep reactive_attrs if the row carries its own token: the swapped row signs the row class's real name, so its next action verifies as usual.
• Un-pending is the settle's job. Only a settle that replaces (s.replace) or removes (s.remove) the row undoes the markup; a flash-only settle, finish: true or a failed job clears the markers only.

Every one routes through the same Phlex::Reactive::Collections decisions the actor's reply uses — the 0↔1 empty-state boundary and the count companion live in exactly one place, so the job-side and reply-side paths cannot drift.

The handle rides ActiveJob metadata, not perform's arity. reply.pending installs it in a thread-local and runs your enqueue inside it; Settles#initialize captures it onto the job instance and #serialize copies it into the job's metadata. So every other caller of the same job — a nightly sweep, a webhook — keeps working unchanged, and reactive_settle is simply a no-op there. That is load-bearing: these jobs almost always have non-UI callers.

It works under enqueue_after_transaction_commit = true. Rails defers that enqueue to ActiveRecord.after_all_transactions_commit, and the endpoint runs your action inside a transaction — so the enqueue (and its serialize) happens after the block has exited. The capture is therefore at job instantiation, which is synchronous inside the block either way. Nothing to configure.

A job that raises still clears the pending state — when it can attribute it. The job:/args: form enqueues one job per record and narrows each job's handle to that record's target, so a failure clears exactly its row and the container, then re-raises for your retry policy. The block form cannot be narrowed (the gem cannot map an arbitrary enqueue back to a record), so a failure there clears nothing and logs why — un-dimming 176 rows that are still working would be a worse lie; finish: true sweeps them up. The subscription is deliberately not torn down on failure — a retry must still be able to reach the actor.

Finishing clears every target, not just the container. A settle that only flashes emits no row stream, so nothing swaps that row's node — the finish sweep is what removes its markers.

One reply.pending per container, per reply. Every pending segment emits a directive targeting the container's id, and the client keys subscriptions by target — a second call would supersede the first and orphan its jobs, so it raises instead.

Peer delivery is best effort. If a peer broadcast fails after the actor's message already went out, the job is not failed — retrying it would re-run perform and send the actor's settle (and its flash) a second time. The failure is logged instead.

One stream key per reply.pending call. A durable broadcast to a never-seen key creates a real PGMQ table (reclaimed by pgbus's hourly orphan sweep at a 24 h threshold), so a key per record would leave 177 tables sitting for a day and open 177 SSE connections. All N settles share one key and one subscription.

Teardown is therefore explicit. Because the key is shared, tearing down on the first arrival would cut off the other N−1. reactive_settle finishes automatically when reply.pending marked exactly one target; a fan-out passes finish: true from whatever knows it is last — a Pgbus::Batch on_finish callback, or the final job of a staggered sequence. Until then the subscription is superseded by the container's next reply.pending, or closed when the page unloads.

The signature is not authorization. reply.pending signs the container's identity so the job can rebuild it off the request thread. That is not permission to act — authorize! in the action, exactly as everywhere else.

broadcast_to(append:) emits the bare row — it has no container instance, so it cannot resolve the declaration or run the size resolver. Its collection counterpart does:

ReconcileQueue.broadcast_collection_to(@bulk_payment, :transfers,
  container: self, in: :unreconcilable, remove: transfer,
  exclude: reactive_connection_id)

row: carries the row component's extra init kwargs — pass the same ones the actor got, or a peer whose row has a required kwarg raises instead of rendering. A remove: may be a record or an already-built dom-id string, matching reply.remove(id, from:).

Row plus count companion plus empty-state toggle. coalesce: (default settle_coalesce_window_ms, 50 ms) applies to the aggregate streams only: the count and empty-state are idempotent replaces of stable targets, so a 177-row fan-out collapses to a handful of them. The row stream is never coalesced — an append is not idempotent, and each row is a distinct target.

Coalescing needs pgbus with zoolutions/pgbus#465; on anything older the window is ignored and every aggregate goes out — chattier, equally correct. The actor's own lane never needs it: its row and aggregates are concatenated into one durable message per settle, so there is nothing extra to collapse.

There is deliberately no settle_token_ttl. A settle has no pull lane for a token to govern — the client cannot poll "is the job done yet", and redeeming such a token at the defer endpoint would render the pre-job component, i.e. the exact bug reply.pending fixes. The settle's wait is bounded by the job.

See also Deferred rendering — the transport this composes with — and Collections for the synchronous add/remove case.