Flow windows not resetting reduce state


I am using Flow to group periodic, incoming events into time-based buckets.

My pipeline and some of my assumptions look like this

def window(interval) do
  Flow.Window.fixed(interval, :second, fn candle -> event.updated_at |> DateTime.to_unix  end)
  |> Flow.Window.trigger_periodically(5, :second, :keep)


Producer # emits an event every minute
|> Flow.from_stage 
|> Flow.partition(stages: 6, hash: &hash/1, window: window(300))
|> Flow.reduce(&empty_bucket/0, fn event, acc ->
  Bucket.merge(acc, event)
|> Flow.each_state(fn bucket_so_far -> send_to_websockets(bucket_so_far) end) # I expect this to be invoked every 5 seconds through the periodic trigger
|> Flow.emit(:state) # To my understanding, this is invoked once per window, not per trigger.
|> Flow.each(fn finished_bucket -> save_bucket_to_db(finished_bucket) end) # I expect this to be invoked every 300 seconds with a 'finished' bucket. I also expect the reducer to reset the bucket state for the next interval.

Problem is that the reduce state never gets reset. I thought it would be reset for each window, but this does not seem to be the case.

Have I misunderstood how fixed windows work?