Storage changes subscription

[tomusdrw]

• Exposing storage changes via RPC pub-sub
• Allows one to subscribe to particular storage keys

CC @jacogr

[jacogr]

Great. Maybe it is better to have the "old" as the alias and the "new" one as the name?

(e.g. chain_subscribeNewHead is actually probably the preferred one to match since it aligns with other RPCs, my gut tells me the preferred one should be the "default")

[dvdplm]

Good stuff. I can't say I understand it all, but overall very readable and interesting.

[dvdplm]

typo: …as keys should be …as key

[tomusdrw]

Rephrased the whole sentence, hope it's clearer now.

[dvdplm]

Couldn't we emit a re-org event with some meta data about what changed and let interested subscribers re-fetch?

[tomusdrw]

Yes, that's possible, although it's not easy to get storage changes for blocks that are already imported/executed. Re-fetching changes would mean to re-execute the blocks, but I suppose based on the filter_keys we could just return all the storage values in the re-orged blocks.

[dvdplm]

Couldn't you check for !susbscribers.is_empty() here? Or is it faster to do it this way?

[tomusdrw]

I'm actually interested only in subscribers for that particular key. So if there is a set of changes:

[(1, Some(2)), (2, None), (3, Some(4))]

but we have no wildcard_listeners and only listener for key=1, changes vector will only contain [(StorageKey(1), Some(StorageData(2))]

[dvdplm]

If I read this right, if the subscriber is gone when we get here, then we assume they've been removed properly already, so we're returning false to avoid calling remove_subscriber() again for them? It's a bit unclear to me how they can still be in the subscribers collection though, can you elaborate on that?

[tomusdrw]

Indeed, I could actually .expect() here, since if the structure is consistent the subscribers should always be in self.sinks. The check here is superfluous.
I could refactor to:

let&(ref sink,ref filter) = self.sinks.get(&subscriber).expect("subscribers returned from self.listeners are always in self.sinks; qed");let result = sink.unbounded_send((hash.clone(),StorageChangeSet{changes: changes.clone(),filter: filter.clone(),}));if result.is_err(){self.remove_subscriber(subscriber);}

or

ifletSome(&(ref sink,ref filter)) = matchself.sinks.get(&subscriber){
let result = sink.unbounded_send((hash.clone(),StorageChangeSet{changes: changes.clone(),filter: filter.clone(),}));if result.is_err(){self.remove_subscriber(subscriber);}}

Which one do you prefer?

[tomusdrw]

Oh, actually I can't since .get() borrows immutably, so remove_subscriber has to be outside of the scope.

[dvdplm]

The channels are closed here, correct?

[tomusdrw]

Yes and since the receiving end is dropped, sending to such channel will trigger an error.

[dvdplm]

…is over…? Do you mean …is closed?

[tomusdrw]

is over as in is finished/is done. Which means that the stream will not emit any more items.

[dvdplm]

Is "prospective" sort of like "pending"?

[tomusdrw]

Yes, those are changes that can still be easily discarded. You can see example of usage inside block builder:

1. We run a transaction
2. It produces a set of prospective changes
3. If we detect that it's somehow invalid we discard the prospective changes
4. If we accept the transaction we commit prospective changes.

[svyatonik]

Are you sure that it should be called before transaction is committed?

[tomusdrw]

Good point. Moved the notification after commit and also guarded by the same if as block import notification.

[gavofyork]

Aside from the minor comment

[gavofyork]

Substrate, not Polkadot :)

[tomusdrw]

Fixed.

[tomusdrw]

• Exposing storage changes via RPC pub-sub
• Allows one to subscribe to particular storage keys

CC @jacogr

[jacogr]

Great. Maybe it is better to have the "old" as the alias and the "new" one as the name?

(e.g. chain_subscribeNewHead is actually probably the preferred one to match since it aligns with other RPCs, my gut tells me the preferred one should be the "default")

[dvdplm]

Good stuff. I can't say I understand it all, but overall very readable and interesting.

[dvdplm]

typo: …as keys should be …as key

[tomusdrw]

Rephrased the whole sentence, hope it's clearer now.

[dvdplm]

Couldn't we emit a re-org event with some meta data about what changed and let interested subscribers re-fetch?

[tomusdrw]

Yes, that's possible, although it's not easy to get storage changes for blocks that are already imported/executed. Re-fetching changes would mean to re-execute the blocks, but I suppose based on the filter_keys we could just return all the storage values in the re-orged blocks.

[dvdplm]

Couldn't you check for !susbscribers.is_empty() here? Or is it faster to do it this way?

[tomusdrw]

I'm actually interested only in subscribers for that particular key. So if there is a set of changes:

[(1, Some(2)), (2, None), (3, Some(4))]

but we have no wildcard_listeners and only listener for key=1, changes vector will only contain [(StorageKey(1), Some(StorageData(2))]

[dvdplm]

If I read this right, if the subscriber is gone when we get here, then we assume they've been removed properly already, so we're returning false to avoid calling remove_subscriber() again for them? It's a bit unclear to me how they can still be in the subscribers collection though, can you elaborate on that?

[tomusdrw]

Indeed, I could actually .expect() here, since if the structure is consistent the subscribers should always be in self.sinks. The check here is superfluous.
I could refactor to:

let&(ref sink,ref filter) = self.sinks.get(&subscriber).expect("subscribers returned from self.listeners are always in self.sinks; qed");let result = sink.unbounded_send((hash.clone(),StorageChangeSet{changes: changes.clone(),filter: filter.clone(),}));if result.is_err(){self.remove_subscriber(subscriber);}

or

ifletSome(&(ref sink,ref filter)) = matchself.sinks.get(&subscriber){
let result = sink.unbounded_send((hash.clone(),StorageChangeSet{changes: changes.clone(),filter: filter.clone(),}));if result.is_err(){self.remove_subscriber(subscriber);}}

Which one do you prefer?

[tomusdrw]

Oh, actually I can't since .get() borrows immutably, so remove_subscriber has to be outside of the scope.

[dvdplm]

The channels are closed here, correct?

[tomusdrw]

Yes and since the receiving end is dropped, sending to such channel will trigger an error.

[dvdplm]

…is over…? Do you mean …is closed?

[tomusdrw]

is over as in is finished/is done. Which means that the stream will not emit any more items.

[dvdplm]

Is "prospective" sort of like "pending"?

[tomusdrw]

Yes, those are changes that can still be easily discarded. You can see example of usage inside block builder:

1. We run a transaction
2. It produces a set of prospective changes
3. If we detect that it's somehow invalid we discard the prospective changes
4. If we accept the transaction we commit prospective changes.

[svyatonik]

Are you sure that it should be called before transaction is committed?

[tomusdrw]

Good point. Moved the notification after commit and also guarded by the same if as block import notification.

[gavofyork]

Aside from the minor comment

[gavofyork]

Substrate, not Polkadot :)

[tomusdrw]

Fixed.

[tomusdrw]

• Exposing storage changes via RPC pub-sub
• Allows one to subscribe to particular storage keys

CC @jacogr

[jacogr]

Great. Maybe it is better to have the "old" as the alias and the "new" one as the name?

(e.g. chain_subscribeNewHead is actually probably the preferred one to match since it aligns with other RPCs, my gut tells me the preferred one should be the "default")

[dvdplm]

Good stuff. I can't say I understand it all, but overall very readable and interesting.

[dvdplm]

typo: …as keys should be …as key

[tomusdrw]

Rephrased the whole sentence, hope it's clearer now.

[dvdplm]

Couldn't we emit a re-org event with some meta data about what changed and let interested subscribers re-fetch?

[tomusdrw]

Yes, that's possible, although it's not easy to get storage changes for blocks that are already imported/executed. Re-fetching changes would mean to re-execute the blocks, but I suppose based on the filter_keys we could just return all the storage values in the re-orged blocks.

[dvdplm]

Couldn't you check for !susbscribers.is_empty() here? Or is it faster to do it this way?

[tomusdrw]

I'm actually interested only in subscribers for that particular key. So if there is a set of changes:

[(1, Some(2)), (2, None), (3, Some(4))]

but we have no wildcard_listeners and only listener for key=1, changes vector will only contain [(StorageKey(1), Some(StorageData(2))]

[dvdplm]

If I read this right, if the subscriber is gone when we get here, then we assume they've been removed properly already, so we're returning false to avoid calling remove_subscriber() again for them? It's a bit unclear to me how they can still be in the subscribers collection though, can you elaborate on that?

[tomusdrw]

Indeed, I could actually .expect() here, since if the structure is consistent the subscribers should always be in self.sinks. The check here is superfluous.
I could refactor to:

let&(ref sink,ref filter) = self.sinks.get(&subscriber).expect("subscribers returned from self.listeners are always in self.sinks; qed");let result = sink.unbounded_send((hash.clone(),StorageChangeSet{changes: changes.clone(),filter: filter.clone(),}));if result.is_err(){self.remove_subscriber(subscriber);}

or

ifletSome(&(ref sink,ref filter)) = matchself.sinks.get(&subscriber){
let result = sink.unbounded_send((hash.clone(),StorageChangeSet{changes: changes.clone(),filter: filter.clone(),}));if result.is_err(){self.remove_subscriber(subscriber);}}

Which one do you prefer?

[tomusdrw]

Oh, actually I can't since .get() borrows immutably, so remove_subscriber has to be outside of the scope.

[dvdplm]

The channels are closed here, correct?

[tomusdrw]

Yes and since the receiving end is dropped, sending to such channel will trigger an error.

[dvdplm]

…is over…? Do you mean …is closed?

[tomusdrw]

is over as in is finished/is done. Which means that the stream will not emit any more items.

[dvdplm]

Is "prospective" sort of like "pending"?

[tomusdrw]

Yes, those are changes that can still be easily discarded. You can see example of usage inside block builder:

1. We run a transaction
2. It produces a set of prospective changes
3. If we detect that it's somehow invalid we discard the prospective changes
4. If we accept the transaction we commit prospective changes.

[svyatonik]

Are you sure that it should be called before transaction is committed?

[tomusdrw]

Good point. Moved the notification after commit and also guarded by the same if as block import notification.

[gavofyork]

Aside from the minor comment

[gavofyork]

Substrate, not Polkadot :)

[tomusdrw]

Fixed.

[tomusdrw]

• Exposing storage changes via RPC pub-sub
• Allows one to subscribe to particular storage keys

CC @jacogr

[jacogr]

Great. Maybe it is better to have the "old" as the alias and the "new" one as the name?

(e.g. chain_subscribeNewHead is actually probably the preferred one to match since it aligns with other RPCs, my gut tells me the preferred one should be the "default")

[dvdplm]

Good stuff. I can't say I understand it all, but overall very readable and interesting.

[dvdplm]

typo: …as keys should be …as key

[tomusdrw]

Rephrased the whole sentence, hope it's clearer now.

[dvdplm]

Couldn't we emit a re-org event with some meta data about what changed and let interested subscribers re-fetch?

[tomusdrw]

Yes, that's possible, although it's not easy to get storage changes for blocks that are already imported/executed. Re-fetching changes would mean to re-execute the blocks, but I suppose based on the filter_keys we could just return all the storage values in the re-orged blocks.

[dvdplm]

Couldn't you check for !susbscribers.is_empty() here? Or is it faster to do it this way?

[tomusdrw]

I'm actually interested only in subscribers for that particular key. So if there is a set of changes:

[(1, Some(2)), (2, None), (3, Some(4))]

but we have no wildcard_listeners and only listener for key=1, changes vector will only contain [(StorageKey(1), Some(StorageData(2))]

[dvdplm]

If I read this right, if the subscriber is gone when we get here, then we assume they've been removed properly already, so we're returning false to avoid calling remove_subscriber() again for them? It's a bit unclear to me how they can still be in the subscribers collection though, can you elaborate on that?

[tomusdrw]

Indeed, I could actually .expect() here, since if the structure is consistent the subscribers should always be in self.sinks. The check here is superfluous.
I could refactor to:

let&(ref sink,ref filter) = self.sinks.get(&subscriber).expect("subscribers returned from self.listeners are always in self.sinks; qed");let result = sink.unbounded_send((hash.clone(),StorageChangeSet{changes: changes.clone(),filter: filter.clone(),}));if result.is_err(){self.remove_subscriber(subscriber);}

or

ifletSome(&(ref sink,ref filter)) = matchself.sinks.get(&subscriber){
let result = sink.unbounded_send((hash.clone(),StorageChangeSet{changes: changes.clone(),filter: filter.clone(),}));if result.is_err(){self.remove_subscriber(subscriber);}}

Which one do you prefer?

[tomusdrw]

Oh, actually I can't since .get() borrows immutably, so remove_subscriber has to be outside of the scope.

[dvdplm]

The channels are closed here, correct?

[tomusdrw]

Yes and since the receiving end is dropped, sending to such channel will trigger an error.

[dvdplm]

…is over…? Do you mean …is closed?

[tomusdrw]

is over as in is finished/is done. Which means that the stream will not emit any more items.

[dvdplm]

Is "prospective" sort of like "pending"?

[tomusdrw]

Yes, those are changes that can still be easily discarded. You can see example of usage inside block builder:

1. We run a transaction
2. It produces a set of prospective changes
3. If we detect that it's somehow invalid we discard the prospective changes
4. If we accept the transaction we commit prospective changes.

[svyatonik]

Are you sure that it should be called before transaction is committed?

[tomusdrw]

Good point. Moved the notification after commit and also guarded by the same if as block import notification.

[gavofyork]

Aside from the minor comment

[gavofyork]

Substrate, not Polkadot :)

[tomusdrw]

Fixed.

[tomusdrw]

• Exposing storage changes via RPC pub-sub
• Allows one to subscribe to particular storage keys

CC @jacogr

[jacogr]

Great. Maybe it is better to have the "old" as the alias and the "new" one as the name?

(e.g. chain_subscribeNewHead is actually probably the preferred one to match since it aligns with other RPCs, my gut tells me the preferred one should be the "default")

[dvdplm]

Good stuff. I can't say I understand it all, but overall very readable and interesting.

[dvdplm]

typo: …as keys should be …as key

[tomusdrw]

Rephrased the whole sentence, hope it's clearer now.

[dvdplm]

Couldn't we emit a re-org event with some meta data about what changed and let interested subscribers re-fetch?

[tomusdrw]

Yes, that's possible, although it's not easy to get storage changes for blocks that are already imported/executed. Re-fetching changes would mean to re-execute the blocks, but I suppose based on the filter_keys we could just return all the storage values in the re-orged blocks.

[dvdplm]

Couldn't you check for !susbscribers.is_empty() here? Or is it faster to do it this way?

[tomusdrw]

I'm actually interested only in subscribers for that particular key. So if there is a set of changes:

[(1, Some(2)), (2, None), (3, Some(4))]

but we have no wildcard_listeners and only listener for key=1, changes vector will only contain [(StorageKey(1), Some(StorageData(2))]

[dvdplm]

If I read this right, if the subscriber is gone when we get here, then we assume they've been removed properly already, so we're returning false to avoid calling remove_subscriber() again for them? It's a bit unclear to me how they can still be in the subscribers collection though, can you elaborate on that?

[tomusdrw]

Indeed, I could actually .expect() here, since if the structure is consistent the subscribers should always be in self.sinks. The check here is superfluous.
I could refactor to:

let&(ref sink,ref filter) = self.sinks.get(&subscriber).expect("subscribers returned from self.listeners are always in self.sinks; qed");let result = sink.unbounded_send((hash.clone(),StorageChangeSet{changes: changes.clone(),filter: filter.clone(),}));if result.is_err(){self.remove_subscriber(subscriber);}

or

ifletSome(&(ref sink,ref filter)) = matchself.sinks.get(&subscriber){
let result = sink.unbounded_send((hash.clone(),StorageChangeSet{changes: changes.clone(),filter: filter.clone(),}));if result.is_err(){self.remove_subscriber(subscriber);}}

Which one do you prefer?

[tomusdrw]

Oh, actually I can't since .get() borrows immutably, so remove_subscriber has to be outside of the scope.

[dvdplm]

The channels are closed here, correct?

[tomusdrw]

Yes and since the receiving end is dropped, sending to such channel will trigger an error.

[dvdplm]

…is over…? Do you mean …is closed?

[tomusdrw]

is over as in is finished/is done. Which means that the stream will not emit any more items.

[dvdplm]

Is "prospective" sort of like "pending"?

[tomusdrw]

Yes, those are changes that can still be easily discarded. You can see example of usage inside block builder:

1. We run a transaction
2. It produces a set of prospective changes
3. If we detect that it's somehow invalid we discard the prospective changes
4. If we accept the transaction we commit prospective changes.

[svyatonik]

Are you sure that it should be called before transaction is committed?

[tomusdrw]

Good point. Moved the notification after commit and also guarded by the same if as block import notification.

[gavofyork]

Aside from the minor comment

[gavofyork]

Substrate, not Polkadot :)

[tomusdrw]

Fixed.

[tomusdrw]

• Exposing storage changes via RPC pub-sub
• Allows one to subscribe to particular storage keys

CC @jacogr

[jacogr]

Great. Maybe it is better to have the "old" as the alias and the "new" one as the name?

(e.g. chain_subscribeNewHead is actually probably the preferred one to match since it aligns with other RPCs, my gut tells me the preferred one should be the "default")

[dvdplm]

Good stuff. I can't say I understand it all, but overall very readable and interesting.

[dvdplm]

typo: …as keys should be …as key

[tomusdrw]

Rephrased the whole sentence, hope it's clearer now.

[dvdplm]

Couldn't we emit a re-org event with some meta data about what changed and let interested subscribers re-fetch?

[tomusdrw]

Yes, that's possible, although it's not easy to get storage changes for blocks that are already imported/executed. Re-fetching changes would mean to re-execute the blocks, but I suppose based on the filter_keys we could just return all the storage values in the re-orged blocks.

[dvdplm]

Couldn't you check for !susbscribers.is_empty() here? Or is it faster to do it this way?

[tomusdrw]

I'm actually interested only in subscribers for that particular key. So if there is a set of changes:

[(1, Some(2)), (2, None), (3, Some(4))]

but we have no wildcard_listeners and only listener for key=1, changes vector will only contain [(StorageKey(1), Some(StorageData(2))]

[dvdplm]

If I read this right, if the subscriber is gone when we get here, then we assume they've been removed properly already, so we're returning false to avoid calling remove_subscriber() again for them? It's a bit unclear to me how they can still be in the subscribers collection though, can you elaborate on that?

[tomusdrw]

Indeed, I could actually .expect() here, since if the structure is consistent the subscribers should always be in self.sinks. The check here is superfluous.
I could refactor to:

let&(ref sink,ref filter) = self.sinks.get(&subscriber).expect("subscribers returned from self.listeners are always in self.sinks; qed");let result = sink.unbounded_send((hash.clone(),StorageChangeSet{changes: changes.clone(),filter: filter.clone(),}));if result.is_err(){self.remove_subscriber(subscriber);}

or

ifletSome(&(ref sink,ref filter)) = matchself.sinks.get(&subscriber){
let result = sink.unbounded_send((hash.clone(),StorageChangeSet{changes: changes.clone(),filter: filter.clone(),}));if result.is_err(){self.remove_subscriber(subscriber);}}

Which one do you prefer?

[tomusdrw]

Oh, actually I can't since .get() borrows immutably, so remove_subscriber has to be outside of the scope.

[dvdplm]

The channels are closed here, correct?

[tomusdrw]

Yes and since the receiving end is dropped, sending to such channel will trigger an error.

[dvdplm]

…is over…? Do you mean …is closed?

[tomusdrw]

is over as in is finished/is done. Which means that the stream will not emit any more items.

[dvdplm]

Is "prospective" sort of like "pending"?

[tomusdrw]

Yes, those are changes that can still be easily discarded. You can see example of usage inside block builder:

1. We run a transaction
2. It produces a set of prospective changes
3. If we detect that it's somehow invalid we discard the prospective changes
4. If we accept the transaction we commit prospective changes.

[svyatonik]

Are you sure that it should be called before transaction is committed?

[tomusdrw]

Good point. Moved the notification after commit and also guarded by the same if as block import notification.

[gavofyork]

Aside from the minor comment

[gavofyork]

Substrate, not Polkadot :)

[tomusdrw]

Fixed.

[tomusdrw]

• Exposing storage changes via RPC pub-sub
• Allows one to subscribe to particular storage keys

CC @jacogr

[jacogr]

Great. Maybe it is better to have the "old" as the alias and the "new" one as the name?

(e.g. chain_subscribeNewHead is actually probably the preferred one to match since it aligns with other RPCs, my gut tells me the preferred one should be the "default")

[dvdplm]

Good stuff. I can't say I understand it all, but overall very readable and interesting.

[dvdplm]

typo: …as keys should be …as key

[tomusdrw]

Rephrased the whole sentence, hope it's clearer now.

[dvdplm]

Couldn't we emit a re-org event with some meta data about what changed and let interested subscribers re-fetch?

[tomusdrw]

Yes, that's possible, although it's not easy to get storage changes for blocks that are already imported/executed. Re-fetching changes would mean to re-execute the blocks, but I suppose based on the filter_keys we could just return all the storage values in the re-orged blocks.

[dvdplm]

Couldn't you check for !susbscribers.is_empty() here? Or is it faster to do it this way?

[tomusdrw]

I'm actually interested only in subscribers for that particular key. So if there is a set of changes:

[(1, Some(2)), (2, None), (3, Some(4))]

but we have no wildcard_listeners and only listener for key=1, changes vector will only contain [(StorageKey(1), Some(StorageData(2))]

[dvdplm]

If I read this right, if the subscriber is gone when we get here, then we assume they've been removed properly already, so we're returning false to avoid calling remove_subscriber() again for them? It's a bit unclear to me how they can still be in the subscribers collection though, can you elaborate on that?

[tomusdrw]

Indeed, I could actually .expect() here, since if the structure is consistent the subscribers should always be in self.sinks. The check here is superfluous.
I could refactor to:

let&(ref sink,ref filter) = self.sinks.get(&subscriber).expect("subscribers returned from self.listeners are always in self.sinks; qed");let result = sink.unbounded_send((hash.clone(),StorageChangeSet{changes: changes.clone(),filter: filter.clone(),}));if result.is_err(){self.remove_subscriber(subscriber);}

or

ifletSome(&(ref sink,ref filter)) = matchself.sinks.get(&subscriber){
let result = sink.unbounded_send((hash.clone(),StorageChangeSet{changes: changes.clone(),filter: filter.clone(),}));if result.is_err(){self.remove_subscriber(subscriber);}}

Which one do you prefer?

[tomusdrw]

Oh, actually I can't since .get() borrows immutably, so remove_subscriber has to be outside of the scope.

[dvdplm]

The channels are closed here, correct?

[tomusdrw]

Yes and since the receiving end is dropped, sending to such channel will trigger an error.

[dvdplm]

…is over…? Do you mean …is closed?

[tomusdrw]

is over as in is finished/is done. Which means that the stream will not emit any more items.

[dvdplm]

Is "prospective" sort of like "pending"?

[tomusdrw]

Yes, those are changes that can still be easily discarded. You can see example of usage inside block builder:

1. We run a transaction
2. It produces a set of prospective changes
3. If we detect that it's somehow invalid we discard the prospective changes
4. If we accept the transaction we commit prospective changes.

[svyatonik]

Are you sure that it should be called before transaction is committed?

[tomusdrw]

Good point. Moved the notification after commit and also guarded by the same if as block import notification.

[gavofyork]

Aside from the minor comment

[gavofyork]

Substrate, not Polkadot :)

[tomusdrw]

Fixed.

[tomusdrw]

• Exposing storage changes via RPC pub-sub
• Allows one to subscribe to particular storage keys

CC @jacogr

[jacogr]

Great. Maybe it is better to have the "old" as the alias and the "new" one as the name?

(e.g. chain_subscribeNewHead is actually probably the preferred one to match since it aligns with other RPCs, my gut tells me the preferred one should be the "default")

[dvdplm]

Good stuff. I can't say I understand it all, but overall very readable and interesting.

[dvdplm]

typo: …as keys should be …as key

[tomusdrw]

Rephrased the whole sentence, hope it's clearer now.

[dvdplm]

Couldn't we emit a re-org event with some meta data about what changed and let interested subscribers re-fetch?

[tomusdrw]

Yes, that's possible, although it's not easy to get storage changes for blocks that are already imported/executed. Re-fetching changes would mean to re-execute the blocks, but I suppose based on the filter_keys we could just return all the storage values in the re-orged blocks.

[dvdplm]

Couldn't you check for !susbscribers.is_empty() here? Or is it faster to do it this way?

[tomusdrw]

I'm actually interested only in subscribers for that particular key. So if there is a set of changes:

[(1, Some(2)), (2, None), (3, Some(4))]

but we have no wildcard_listeners and only listener for key=1, changes vector will only contain [(StorageKey(1), Some(StorageData(2))]

[dvdplm]

If I read this right, if the subscriber is gone when we get here, then we assume they've been removed properly already, so we're returning false to avoid calling remove_subscriber() again for them? It's a bit unclear to me how they can still be in the subscribers collection though, can you elaborate on that?

[tomusdrw]

Indeed, I could actually .expect() here, since if the structure is consistent the subscribers should always be in self.sinks. The check here is superfluous.
I could refactor to:

let&(ref sink,ref filter) = self.sinks.get(&subscriber).expect("subscribers returned from self.listeners are always in self.sinks; qed");let result = sink.unbounded_send((hash.clone(),StorageChangeSet{changes: changes.clone(),filter: filter.clone(),}));if result.is_err(){self.remove_subscriber(subscriber);}

or

ifletSome(&(ref sink,ref filter)) = matchself.sinks.get(&subscriber){
let result = sink.unbounded_send((hash.clone(),StorageChangeSet{changes: changes.clone(),filter: filter.clone(),}));if result.is_err(){self.remove_subscriber(subscriber);}}

Which one do you prefer?

[tomusdrw]

Oh, actually I can't since .get() borrows immutably, so remove_subscriber has to be outside of the scope.

[dvdplm]

The channels are closed here, correct?

[tomusdrw]

Yes and since the receiving end is dropped, sending to such channel will trigger an error.

[dvdplm]

…is over…? Do you mean …is closed?

[tomusdrw]

is over as in is finished/is done. Which means that the stream will not emit any more items.

[dvdplm]

Is "prospective" sort of like "pending"?

[tomusdrw]

Yes, those are changes that can still be easily discarded. You can see example of usage inside block builder:

1. We run a transaction
2. It produces a set of prospective changes
3. If we detect that it's somehow invalid we discard the prospective changes
4. If we accept the transaction we commit prospective changes.

[svyatonik]

Are you sure that it should be called before transaction is committed?

[tomusdrw]

Good point. Moved the notification after commit and also guarded by the same if as block import notification.

[gavofyork]

Aside from the minor comment

[gavofyork]

Substrate, not Polkadot :)

[tomusdrw]

Fixed.