add

Registers subscription. On a running runtime it starts at once in every partition: it catches up from its checkpoint (or as Subscription.startFrom says) on its own read, then joins the shared tail.