* Rework receiver restart. - Protect against concurrent restarts - Clean up and consolidate code around restarts - Use `RestartStream` of buffer rather than creating new buffers. * fix test