es_sink: fix to ensure we do not consume beyond any running or queued events