Store stream ended state in the state of the aggregator

To be used to garbage-collect stale state.
Avoid creating a 'processed' entry if there is no way to process the file