class BroadcastAll_Rewrite
  include Bud

  state do
    channel :chn, [:@addr, :id] => [:val]
    channel :chn_ack, [:@rce_sender, :addr, :id]
    range :chn_approx, [:addr, :id]
    table :log, [:id] => [:val]
    sealed :node, [:addr]
    scratch :r0_node_log_joinbuf, [:node_addr, :log_id, :log_val]
    scratch :r0_node_log_missing, [:node_addr, :log_id, :log_val]
    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 < -(log * seal_node).lefts.notin(r0_node_log_missing, :id => :log_id, :val => :log_val))
    (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)
    log <= (chn.payloads)
    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)
  end
end