You might want to look at what DDflow [0] can do, in regards to handling emission/retraction.

If you figure out how this would work in the context of DDflow, let us know. This sounds very interesting.

[0]: https://github.com/TimelyDataflow/differential-dataflow