class BroadcastEpochAll_Rewrite
  include Bud

  state do
    channel :chn, [:@addr, :id] => [:epoch, :val]
    channel :chn_ack, [:@rce_sender, :addr, :id]
    range :chn_approx, [:addr, :id]
    scratch :del_log_r0, [:id] => [:epoch, :val]
    scratch :del_node_r0, [:addr, :epoch]
    table :log, [:id] => [:epoch, :val]
    table :node, [:addr, :epoch]
    scratch :r0_node_log_joinbuf, [:node_addr, :node_epoch, :log_id, :log_epoch, :log_val]
    scratch :r0_node_log_missing, [:node_addr, :node_epoch, :log_id, :log_epoch, :log_val]
    range :seal_log, [:ignored]
    range :seal_log_epoch, [:epoch]
    range :seal_node, [:ignored]
    range :seal_node_epoch, [:epoch]
  end

  bloom do
    (chn < (node * log).pairs(:epoch => :epoch) { |n, l| ([n.addr] + l) }.notin(chn_approx, 0 => :addr, 1 => :id).~)
    (log < -del_log_r0)
    (node < -del_node_r0)
    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, :epoch => :log_epoch, :val => :log_val)
    del_log_r0 <= (log * seal_node_epoch).lefts(:epoch => :epoch).notin(r0_node_log_missing, :id => :log_id, :epoch => :log_epoch, :val => :log_val)
    del_node_r0 <= (node * seal_log).lefts.notin(r0_node_log_missing, :addr => :node_addr, :epoch => :node_epoch)
    del_node_r0 <= (node * seal_log_epoch).lefts(:epoch => :epoch).notin(r0_node_log_missing, :addr => :node_addr, :epoch => :node_epoch)
    log <= (chn.payloads)
    r0_node_log_joinbuf <= (node * log * chn_approx).combos(node.epoch => log.epoch, chn_approx.addr => node.addr, chn_approx.id => log.id) {|x,y,z| x + y}
    r0_node_log_missing <= (node * log).pairs(node.epoch => log.epoch) {|x,y| x + y}.notin(r0_node_log_joinbuf)
  end
end