Finer grained event subscription API

[ascjones]

Currently the Event subscription API attempts to simplify subscribing to events, at the cost of not giving the user very much control over which events are received from the subscription, and making some assumptions which may be false. See #310, #307, #283, #306 and maybe more.

We should consider an API similar to polkadot.js, which will give the user full access to the transaction status etc.

// Make a transfer from Alice to BOB, waiting for inclusion
const unsub = await api.tx.balances
  .transfer(BOB, 12345)
  .signAndSend(alice, (result) => {
    console.log(`Current status is ${result.status}`);

    if (result.status.isInBlock) {
      console.log(`Transaction included at blockHash ${result.status.asInBlock}`);
    } else if (result.status.isFinalized) {
      console.log(`Transaction finalized at blockHash ${result.status.asFinalized}`);
      unsub();
    }
  });

In this way the user receives all events, and is responsible for filtering and acting upon them.

One obvious thing we need to consider is how to handle subscription cancellation ergonomically in Rust with async/await.

See also #309 (comment)

[jsdw]

So, the things that might be useful to know when you subscribe to a given extrinsic submission are, I think:

• How is the extrinsic progressing (the TransactionStatus eg Ready, InBlock, Finalized..).
• When it's in a block:
  • which block is it in,
  • what is its index in the block
  • the events that are associated with it (these can be obtained from the block hash/ext ID above).

And in general, if you want to subscribe to events, you might want to know the current block hash, the Phase (ie what produced the event) and the event data (probably pre-parsed; something like RawEvent; potentially we could lean on metadata a little more to save some allocations).

For the submit-and-subscribe to extrinsic use case, would an API that works a bit like this work?:

// Submit extrinsic. Return stream denoting progress:
let mut progress = api.tx()
    .balances()
    .transfer(dest, 10_000)
    .sign_and_submit_then_watch(&signer)
    .await?;

// This fires each time there is a progress update. `progress` impls Stream.
// this is the "low level" API that lets people do what they want with the progress data.
while let Some(current_status) = progress.next()? {
    
    // current_status looks something like TransactionStatus, but when the extrinsic makes
    // it into a block, we want back the block hash _and_ the extrinsic ID.

    // At the lowest level, we can match on this enum:
    match current_status {
        InBlock(in_block) => {
            println!("In block hash {}, ext id {}", in_block.hash, in_block.extrinsic_id);
        },
        ...
    }

    // Convenience functions exist to match on common states:
    if let Some(finalized) = current_status.as_finalized() {
        println!("Finalized in block hash {}, ext id {}", finalized.hash, finalized.extrinsic_id);
        // Probably some convenience function to get the raw events, too:
        let events: RawEvents = finalized.events().await?;
        for raw_event in &events {
            println!("{:?}", raw_event);
        }

        // One of those events is denotes extrinsic success or failure, which 
        // should also be easily accessible:
        let did_succeed: bool = events.did_succeed();
    }
}

// That lower level API provides access to everything you'd care about but is more verbose.
// We can add higher level calls for common operations, like:

// Wait for ext to appear in finalized block and remark on it's success
let did_succeed: bool = progress
    .wait_for_finalized()
    .await?
    .did_succeed();

// Similar to above, but get the associated events (probably after the await
// we have the same object as `finalized` in the `current_status.as_finalized()` above)..)
let events: RawEvents = progress
    .wait_for_finalized()
    .await?
    .events();

if events.did_succeed() {
    println!("Success!");
}

This boils down to being able to track the transaction status, and get back the block hash and ext ID as appropriate from that. From that, we can get hold of the associated events in a block, which will probably just leans on the general event subscription mechanism.

The general event subscription mechanism has some "hard coded" filters presently (filter_block, filter_extrinsic and filter_event). I'd lean towards having functions to return something equivalent to Stream<Item=Vec<(BlockHash, Phase,RawEvent)>> on it (but likely a nicer named struct for the Item) to give the raw output, and then a potentially more ergonomic flattened version of that (ie Item=(BlockHash, Phase,RawEvent) but again named props; less efficient but easier to work with).

With that, users can rely on StreamExt::filter and such to work with the event stream however they like, and we could provide a raw function that returns the event details for a single block, which can be used to back this, or used independently when you know what block you want events for.

I'm very open to thoughts on all of this; what do you think?

[ascjones]

Yes I like this philosophy of giving full access to transaction status, block phase etc. while also adding convenience methods for common patterns.

The general event subscription mechanism has some "hard coded" filters presently (filter_block, filter_extrinsic and filter_event). I'd lean towards having functions to return something equivalent to Stream<Item=Vec<(BlockHash, Phase,RawEvent)>> on it (but likely a nicer named struct for the Item) to give the raw output, and then a potentially more ergonomic flattened version of that (ie Item=(BlockHash, Phase,RawEvent) but again named props; less efficient but easier to work with).

Agree with this. However if I vaguely remember some difficulty making the subscription into a Stream, however worth giving it a go again since the Futures and Stream stuff has moved on again since then.

I'd also be interested to hear from @gregdhill and @sander2 what they think, since they have used and contributed to the existing API.

[sander2]

I'd also be interested to hear from @gregdhill and @sander2 what they think, since they have used and contributed to the existing API.

Regarding the TransactionStatus updates, I don't think we'd have much use for that in our project beyond adding some debug prints - mostly we only want to act on finalized data. As long as there is a higher-level interface like wait_for_finalized, then I don't think there's any reason not to add it - it will be very nice for projects that have more direct user interaction.

As for the event subscription refactoring, I agree that the current solution is not the most elegant. Having the subscription return a stream could be an improvement in that regard, so that it should be easy to compose the desired filters with filter, filter_map, take_while, etc. Although as usual with streams, the possible errors will probably complicate it somewhat.

[jsdw]

Thanks for the feedback so far! My plan is to leave this open for feedback (particularly from @gregdhill) probably until mid next week depending on my schedule, and if the feedback is positive then I'll start work on implementing the proposed API around then :)

[gregdhill]

I'm quite happy with that API, it seems quite elegant and allows for a lot of flexibility. One related thought for general refactoring, might we want to re-use a single event subscription? If we have a number of concurrent calls that start at the same time they will each setup a websocket subscription with the remote node to read (and then filter) all events.

[jsdw]

That seems like a sensible optimisation to me. I think I'd be inclined to do that if it turns out to be fairly straightforward, and defer it to a separate issue/PR if it turns out to be a more involved change :)

[dvdplm]

Closes #290 too I think?

[ascjones]

Closes #290 too I think?

It does if we replace the ExtrinsicSuccess type with direct filtering on the "stream" of events.

[jsdw]

That is my current intention :)