class BroadcastCausal_Rewrite
  include Bud

  state do
    channel :chn, [:@addr, :id] => [:val, :deps]
    channel :chn_ack, [:@rce_sender, :addr, :id]
    range :chn_approx, [:addr, :id]
    scratch :del_log_r0, [:id] => [:val, :deps]
    scratch :del_log_r2, [:id] => [:val, :deps]
    scratch :flat_dep, [:id, :dep]
    table :log, [:id] => [:val, :deps]
    scratch :missing_dep, [:id, :dep]
    sealed :node, [:addr]
    scratch :pending, [:id] => [:val, :deps]
    scratch :r0_node_log_joinbuf, [:node_addr, :log_id, :log_val, :log_deps]
    scratch :r0_node_log_missing, [:node_addr, :log_id, :log_val, :log_deps]
    table :safe_log, [:id] => [:val, :deps]
    range :seal_log, [:ignored]
    range :seal_node, [:ignored]
  end

  bloom do
    (chn < (node * log).pairs { |n, l| (n + l) }.notin(chn_approx, 0 => :addr, 1 => :id).~)
    (log < -(del_log_r0 * del_log_r2).matches { |t0, t1| t0 })
    (node < -(node * seal_log).lefts.notin(r0_node_log_missing, :addr => :node_addr))
    chn_ack <~ chn {|c| [c.source_addr, c.addr, c.id]}
    chn_approx <= (chn_ack.payloads)
    del_log_r0 <= (log * seal_node).lefts.notin(r0_node_log_missing, :id => :log_id, :val => :log_val, :deps => :log_deps)
    del_log_r2 <= (log * safe_log).lefts(:id => :id)
    flat_dep <= (pending.flat_map { |l| l.deps.pro { |d| [l.id, d] } })
    log <= (chn.payloads)
    missing_dep <= (flat_dep.notin(safe_log, :dep => :id))
    pending <= (log.notin(safe_log, :id => :id))
    r0_node_log_joinbuf <= (node * log * chn_approx).combos(chn_approx.addr => node.addr, chn_approx.id => log.id) {|x,y,z| x + y}
    r0_node_log_missing <= (node * log).pairs {|x,y| x + y}.notin(r0_node_log_joinbuf)
    safe_log < (+pending.notin(missing_dep, :id => :id))
  end
end