Multipublisher Improvements for Manager Internal State

[ata-nas]

Story Form

As a Block Node Engineer/Operator/User
I want to have a properly working Multipublisher plugin
So that I can use the plugin as intended

Technical Notes

• Multiple conversations with some concerns arose recently about the live stream manager and it's internal state which determines how the handlers will behave/respond.
  • For reference, see the following:
    • Validate Bad Proof Possible Race Conditions and Out of Order Acknowledgements #1478
    • This comment and it's reply
• Based on the above, the major concerns are:
  1. Out of Order Acknowledgements:
     • Are out of order acks problematic?
     • What we've most recently discussed based on linked comments above is that an ack means "block 'n' and all previous" and it is not required to reach the publisher strictly in order.
     • Also, the source of the block is irrelevant. If the block node has the blocks, then it should acknowledge to any publisher that might be waiting for that ack.
     • Backfill is permitted to "get ahead" of publishers; the acknowledgement gives the slower publisher a chance to reduce effort and perhaps catch up.
     • Fortunately block proofs cannot be correctly created and verified except in order, so we don't need the publisher plugin to manage the order; the validation requirement ensures we don't get properly out of order.
     • All the above being said however, still could potentially leave room for a miss, imagine block 1 is sent, block 2 is sent, then block 1 fails verification and internal state of the manager is "I expect block 1 next". But then imagine block 2 also fails verification (most likely expected due to failed block 1), then if no one in the meantime has supplied (or started to supply) block 1, the block 2 failed verification notification could overwrite the state of the manager to "expect block 2 next", hence missing block 1.
     • What should/can we do to prevent such misses and do we have an actual concern here?
  2. LiveStreamPublisherManager Internal State Management:
     • The LiveStreamPublisherManager relies on atomic variables to manage it's current state. This determines block actions returned to publishers who query the manager, but also the block item propagation order to the ring buffer.
     • We have three main fields:
       • currentStreamingBlockNumber: this is mainly used in the message forwarder queue where block items are propagated to the ring buffer. It follows the current block number currently being forwarded.
       • nextUnstreamedBlockNumber: this is mainly used to determine what the block action should be for a handler that asks for it. It follows the next expected block number. Once the number starts streaming, this value increments monotonously by 1.
       • lastPersistedBlockNumber: this is also the last known (acknowledged) block. Acknowledgements happen only after a persistence notification is received, because persistence only happens on verified blocks, but also because the block to be persisted is a needed condition for an acknowledgement.
     • We have to rely on the proper state and management of it, of all three values so that we can be absolutely certain that the manager works correctly.
     • We must also ensure that no misses will happen due to the async nature of the manager (example of a miss is stated above with the Out of Order Acknowledgements, where a failure happens just after another failure, where the first one updates the next and current streaming fields to some value, but then the second failure overwrites them before any publisher has had the time to act on the first change)
       • For this purpose we should probably consider to have collections to follow overwrites and then on the two CAS-es, we will have a small calculation to determine what should be the next value.
  3. The handlerIsEnding Event:
     • Similar to above point, we should ensure that we will have no misses here as well.
     • We currently have a check if (blockNumber + 1L == nextUnstreamedBlockNumber.get()) which will miss if the blockNumber is for example not 1 behind but two (blockNumber + 2L == nextUnstreamedBlockNumber.get()). We must verify if that is a valid case (I most likely is) and we need to make sure to capture such misses.
     • Also, currently every single EndStream request has a validation based on proto documentation.
       • Is that correct?
       • Should we consider to still go through the handlerIsEnding logic despite of the request's potential invalidity? We currently rely on the correct request only to potentially trigger an on-demand backfill. The handlerIsEnding logic is triggered based on handler internal state (the block it is currently streaming and if the handler state is reset).
  4. Multiple Blocks Streamed by One Handler Simultaneously:
     • We've recently discussed that a publisher can continue to send block items immediately after finishing a block (documentation update needed)
     • This means that internally, for the manager, we need to allow this to happen, as currently adding the same queue to the queueByBlock map twice will result in a SKIP.
     • Also, we need to make updates to the queue removal logic to support removing the queue from all keys it could be present as a value at, but moreover, if there are multiple blocks in that queue, we need to clear the correct block (this could be tricky)
     • We should consider to simply issue a new queue for each block and supply it to the handler that will stream the block, the handler will only ever stream one block at a time.
       • This also means that in the map where we have the queue by handlerId, we need to update the map to hold a list of queues for a given handler.
       • This also means that when clearing a queue by block number, we can very easily simply remove the queue and drop it's content because we know it will hold only a single block.
• We should validate the above concerns and consider to make changes.

[jsync-swirlds]

This is super long, but I've tried to answer inline:

• Multiple conversations with some concerns arose recently about the live stream manager and it's internal state which determines how the handlers will behave/respond.
  • For reference, see the following:
    • Validate Bad Proof Possible Race Conditions and Out of Order Acknowledgements #1478
    • This comment and it's reply
• Based on the above, the major concerns are:
  1. Out of Order Acknowledgements:
     • Are out of order acks problematic?
       • These should not cause problems.
     • What we've most recently discussed based on linked comments above is that an ack means "block 'n' and all previous" and it is not required to reach the publisher strictly in order.
     • Also, the source of the block is irrelevant. If the block node has the blocks, then it should acknowledge to any publisher that might be waiting for that ack.
     • Backfill is permitted to "get ahead" of publishers; the acknowledgement gives the slower publisher a chance to reduce effort and perhaps catch up.
     • Fortunately block proofs cannot be correctly created (by publishers) and verified except in order, so we don't need the publisher plugin to manage the order; the validation requirement ensures we don't get properly out of order.
       • This isn't strictly true (c.f. backfill), but it should be enforced for the publisher-delivered blocks. We likely need a small amount of change in the verification plugin to ensure the "acked N implies acked N-1" invariant.
     • All the above being said however, still could potentially leave room for a miss, imagine block 1 is sent, block 2 is sent, then block 1 fails verification and internal state of the manager is "I expect block 1 next". But then imagine block 2 also fails verification (most likely expected due to failed block 1), then if no one in the meantime has supplied (or started to supply) block 1, the block 2 failed verification notification could overwrite the state of the manager to "expect block 2 next", hence missing block 1.
       • This won't happen, because we reject "future" failures; we'd just ignore the failure of the "future" block.
     • What should/can we do to prevent such misses and do we have an actual concern here?
  2. LiveStreamPublisherManager Internal State Management:
     • The LiveStreamPublisherManager relies on atomic variables to manage it's current state. This determines block actions returned to publishers who query the manager, but also the block item propagation order to the ring buffer.
     • We have three main fields:
       • currentStreamingBlockNumber: this is mainly used in the message forwarder queue where block items are propagated to the ring buffer. It follows the current block number currently being forwarded.
       • nextUnstreamedBlockNumber: this is mainly used to determine what the block action should be for a handler that asks for it. It follows the next expected block number. Once the number starts streaming, this value increments monotonously by 1 (unless a block fails verification).
       • lastPersistedBlockNumber: this is also the last known (acknowledged) block. Acknowledgements happen only after a persistence notification is received, because persistence only happens on verified blocks, but also because the block to be persisted is a needed condition for an acknowledgement.
     • We have to rely on the proper state and management of it, of all three values so that we can be absolutely certain that the manager works correctly.
     • We must also ensure that no misses will happen due to the async nature of the manager (example of a miss is stated above with the Out of Order Acknowledgements, where a failure happens just after another failure, where the first one updates the next and current streaming fields to some value, but then the second failure overwrites them before any publisher has had the time to act on the first change)
       - Again, the out-of-order failed verification won't break anything.
       • For this purpose we should probably consider to have collections to follow overwrites and then on the two CAS-es, we will have a small calculation to determine what should be the next value.
         • We probably should not do this, it's unnecessary and more likely to create errors.
  3. The handlerIsEnding Event:
     • Similar to above point, we should ensure that we will have no misses here as well.
     • We currently have a check if (blockNumber + 1L == nextUnstreamedBlockNumber.get()) which will miss if the blockNumber is for example not 1 behind but two (blockNumber + 2L == nextUnstreamedBlockNumber.get()). We must verify if that is a valid case (I most likely is) and we need to make sure to capture such misses.
       • This is not a valid case, we're checking if the handler is currently streaming, and that is only true in the exact case currently tested.
     • Also, currently every single EndStream request has a validation based on proto documentation.
       • Is that correct?
         • yes, we must check the EndStream request, a publisher can have bugs, misbehave, or the data could flow through a broken router.
       • Should we consider to still go through the handlerIsEnding logic despite of the request's potential invalidity? We currently rely on the correct request only to potentially trigger an on-demand backfill. The handlerIsEnding logic is triggered based on handler internal state (the block it is currently streaming and if the handler state is reset).
         • Requesting backfill should only happen in restricted cases, and it won't break the world if we miss some rare cases, because the next publisher request will likely catch it.
  4. Multiple Blocks Streamed by One Handler Simultaneously:
     • We've recently discussed that a publisher can continue to send block items immediately after finishing a block (documentation update needed)
     • This means that internally, for the manager, we need to allow this to happen, as currently adding the same queue to the queueByBlock map twice will result in a SKIP.
       • I don't think this is true. I believe this will just add the same queue value for another key (the block numbers are different).
     • Also, we need to make updates to the queue removal logic to support removing the queue from all keys it could be present as a value at, but moreover, if there are multiple blocks in that queue, we need to clear the correct block (this could be tricky)
       • This is valid, but not tricky or difficult (details provided offline).
     • We should consider to simply issue a new queue for each block and supply it to the handler that will stream the block, the handler will only ever stream one block at a time.
       - This is super costly and creates a ton of race conditions and/or blocking requirements.
       • This also means that in the map where we have the queue by handlerId, we need to update the map to hold a list of queues for a given handler.
         • We wouldn't have this map at all, we'd use only the queue-by-block-number map.
       • This also means that when clearing a queue by block number, we can very easily simply remove the queue and drop it's content because we know it will hold only a single block.
         • But we would also have to synchronize to do so.
• We should validate the above concerns and consider to make changes.

[ata-nas]

@jsync-swirlds thanks for your answers! Here my answers to your answers (quoted replies) and my general feedback:

1. Regarding out of order acks:
   • My understanding is that the verification plugin will run verification of blocks in series, not async, so it will not be generally possible to verify block 2 before block 1 because publisher will guarantee to send these items in order, first block 1 and then block 2. Also, the serial verification is something we will stick with. Otherwise this changes things?
   • For backfill, this is a bit different because there we receive the block via notification on the other ring buffer, so that verification can happen async, juxtaposed to the publisher supplied blocks

   • This isn't strictly true (c.f. backfill), but it should be enforced for the publisher-delivered blocks. We likely need a small amount of change in the verification plugin to ensure the "acked N implies acked N-1" invariant.

     • I will make a GH issue for that
   • Still, do we really need no changes in publisher, but only in verification? Is the block source still not relevant for the publisher manager when handling verification notifications in the publisher manager?

   • This won't happen, because we reject "future" failures; we'd just ignore the failure of the "future" block.

     • I cannot seem to wrap my head around this one, based the fact that we will run verification in series, then, yes, it will not happen that for example block 2's failure will be received before block 1's, but still, if not enough time has passed after we have received block 1's failure and no publisher has yet started streaming it again, then block 2's failure will set the internal state of the publisher to expect next to be streamed to be block 2. This skips block 1's resend? Am I missing something, or do you here mean, that the change we need to do for verification mentioned above, where we need to enforce dropping (or not accepting) the next block (last ackd + 2 in this case) when last ackd + 1 fails, until last ackd + 1 is supplied again, will also mean that the notification, that would currently be sent for the verification result of of last ackd + 2, will no longer be sent, so we eliminate the possibility of such conflict (effectively "rejecting future failures")?
   • Lastly, if we are saying that whenever we acknowledge a block, this means that all blocks prior to the just acknowledged block are also acknowledged, including forward backfill? Is that correct? Also, do we need some changes when that is the case? For instance, we talked about that we should acknowledge all blocks, even ones that are from backfill (but still higher than latest ackd) which would reduce the load on a publisher that is running behind. I think we should potentially update the next to stream and current streaming longs as well, currently, we only update latest ackd. This will be relevant only when the persisted notification's newest block number is higher than latest ackd + 1.
     
   • as an action item, I believe we should open GH issue for enforcing publisher-delivered blocks to be handled in order and if a session fails, we enforce to not accept any other block until the failed block is resend, continuing normally afterwards. verification remains serial
   • as another action item, I believe we should open GH issue for persisted notifications that have newest block > latest ackd + 1, to also change next and current streaming longs (manager internal state), not only the latest ackd++
2. Regarding manager internal state management
   • I believe a lot of the concerns here are connected to verification failing and persistence succeeding, which would cause out of order acks, but that should not be the case based on above discussions.
3. RegardingMultiple Blocks Streamed by One Handler Simultaneously:

   • I don't think this is true. I believe this will just add the same queue value for another key (the block numbers are different).

     • Our current implementation assumes that, we even have a break on removal, assuming there will always be a single queue in that map. We need to make amends here if we will support the same value for multiple keys. Below in the pic can be seen why SKIP will be returned, simply if (previousValue != null) will pass, we already have the value present, so entering this if will then skip to the last return: return blockNumber < nextUnstreamedBlockNumber.get() ? BlockAction.SKIP : BlockAction.END_BEHIND;, or, put simply, we assume a handler has "jumped" in front of the caller, which is a correct assumption if we will not allow the same value for multiple keys:
       
       we could have the if check to be if (previousValue != null && previousValue != previousTransferQueueOfHandler) so that if the value of previousValue and previousTransferQueueOfHandler is the same, we know this is not a jump, but attempt to add another queue
     • Similarly, when removing a queue, we need to iterate over all values, not breaking here:
       

     • This is super costly and creates a ton of race conditions and/or blocking requirements.

       • I believe you may be right that there will be too much cost and race conditions where we need to lock for a long time (wile iterating over the whole collection)
     • as an action item I believe we must open a GH issue to add support for multiple usage of the same queue in the queueByBlock map and updating all supporting logic, also to manage proper removal as you have suggested.

EDIT:
For whatever reason, I miss one of the points...

1. Regarding the handlerIsEnding Event:

   • This is not a valid case, we're checking if the handler is currently streaming, and that is only true in the exact case currently tested.

     • Maybe I am not getting this one, when you say we are checking if the handler is currently streaming, do you mean steaming as in it is in the middle of supplying a block, or the message forwarder is currently forwarding that block. Because in theory if it is the former, multiple publishers can be simultaneously "streaming" (i.e. supplying) blocks. I see that any case where the blockNumber is lower than by at least 1 is a valid case. But if you mean the latter, then I believe we must test against currentStreamingBlockNumber not next to stream. Calling this method always means that the block is currently being streamed (if we accept the former definition above), as the publisher only calls this if it is in the middle of the block. Now that I think of it, maybe we should always test against currentStreamingBlockNumber? That depends on the definition above. Also, what I've recently realized is that we should probably also update next/current streaming block numbers? This is still confusing for me.
   • I agree with you that we should validate the requests and not act if the EndStream request is invalid, but we must always close the stream.
   • Now that I think of it, when the handler is shutting down, we should probably update next/current streaming block numbers... I see a possibility where a publisher ends the stream either abruptly or normally, but in the middle of the block, lets say for block 1, we need that block 1 to be supplied again...

[jsync-swirlds]

This is large enough that reply in this tool is impractical.
Let's get somewhere that we can comment on lines instead of repeating the whole comment over and over (we'd end up with book length replies, which are unreadable.
I'll contact you separately for this (also, we probably need a meeting, otherwise this discussion will take months).

[Nana-EC]

@ata-nas to circle back to review now further items to pull out from here.
Then we'll convert to a discussion

[ata-nas]

@ata-nas to circle back to review now further items to pull out from here. Then we'll convert to a discussion

This was resolved elsewhere, here the key takeaways:

• Out of Order Acknowledgements:
  • These are not problematic because they should not be happening.
    • There are two conditions that need to have happened in order to acknowledge a block.
      1. A block must be verified;
      2. A block must be persisted;
    • When these happen, only then can we acknowledge a block.
    • Our current flow is as follows:
      1. Data (block) comes in and is published to block items ring buffer subscribers.
      2. Verification (a subscriber of block items ring buffer) receives the data and can start verifying it.
      3. Data is either verified or fails verification.
         1. If verification passes, then a Verification Notification with success is published to ring buffer for internal messages.
         2. Else, a Verification Notification with failure is published to ring buffer for internal messages.
      4. Assuming verification passed, the live persistence plugin (files.recent) is able to receive the notification as it is subscribed for internal messaging.
      5. Persistence then persists the block, or fails to do so.
         1. If persisted, a Persistence Notification is published.
         2. Else, a Persistence Failed Notification is published (not implemented yet).
      6. Assuming persistence is successful, the Publisher Manager will receive the Persistence Notification, as it is subscribed for internal messaging.
      7. The Publisher Manager can then notify handlers that they need to acknowledge the persisted block.
      8. Similarly, if a failure is detected, handlers will send appropriate responses and the Publisher Manager will properly reset it's state, awaiting the next block.
    • Using the above flow, we can now split the responsibilities properly and conclude that the verification plugin is responsible for sending verification notifications in order.
      • This means that it can asynchronously verify multiple blocks, but it will send verification notifications in block order.
      • During failures this means that the plugin needs to be resupplied the failed block in order to not effectively stall the application (this is the responsibility of the publisher plugin).
      • Just like the publisher plugin can accept multiple blocks at a time async, but then propagates them in order, the verification plugin can accept and verify async multiple blocks, but will be sending verification notifications with result in order.
      • Similarly, persistence will be receiving these in order and will persist them in order and will send Persisted Notifications in order, because each ring buffer subscriber is being notified of events in order, on it's own dedicated thread serially.
  • Using the above, we can conclude that Out of Order Acknowledgements will not be problematic for us.
  • Note, some changes in verification are due in order to ensure that no Out of Order Acknowledgements can happen.
• Publisher Manager Internal State Management:
  • Major concerns here are the proper internal state management of the Publisher Manager.
  • This includes resetting correctly the atomic longs that follow next unstreamed, current streaming, and last acknowledged block. These are critical for the functioning of the plugin.
  • In order to ensure proper state management, we need to ensure that when an unfavorable event happens (failed verification, failed persistence, abrupt handler termination mid block etc. etc.):
    • All streams are closed (or we close only the problematic stream if current state permits it) after the respective publisher has been sent the appropriate response.
    • Publisher Manager Internal State is then properly reset.
    • Publishers are not allowed to publish (effectively change state) until the above is done to completion (this should take very low amount of time, but is critical in order to ensure no race conditions).
  • Because Out of Order Acknowledgements will not happen as seen above, we can conclude that our current logic is sufficient, we only need to resolve some very rare possible race conditions.
  • For both major unfavorable events, Failed Verification && Failed Persistence, we closing all streams, which really simplifies state management.
  • For the unfavorable event of an abrupt termination of a stream from the publisher's side mid block, if current state permits, we will close only it (if it is currently streaming the next unstreamed block -1), else we will close all streams, reset the state properly and request for a resend.
• The handlerIsEnding Event:
  • As mentioned in the above point, a handler (publisher) can end the stream on it's side. When this happens, we expect the EndStream message to be sent to the block node. There are now two possibilities:
    • The publisher is ending the stream right after it publishes the last block in full;
    • The publisher is ending the stream before it publishes the last block in full (mid block);
  • If the former happens, then we are fine, we cleanup resources on our side and close the handler.
  • If the latter happens however, we need to again cleanup resources and close the handler, but we also need to manage some state:
    • Firstly, we need to clear the half finished block from the handler's queue and update the manager's maps.
    • Then, we have two situations:
      • The handler was streaming next unstreamed block -1;
      • The handler was streaming next unstreamed block -2 and below;
    • The former is our happy path, we can simply decrement the next unstreamed block with one and close this and only this handler.
    • The latter requires that we shutdown all handlers, clear all queues and request a resend from the publishers. This is required in order to ensure proper state management and to ensure we will not loose a block, effectively halting the app as other components (mainly verification) require that all blocks arrive sooner or later.
  • Some changes are currently required, but we have everything captured in EPIC: Support multi publisher block stream ingestion #202
• Multiple Blocks Streamed by One Handler Simultaneously:
  • This is permitted.
  • We need to make some changes to fully support that
  • We have captured what we need to do in @todos and EPIC: Support multi publisher block stream ingestion #202 as issues

[ata-nas]

@Nana-EC @jsync-swirlds @mustafauzunn @AlfredoG87 as suggested, converted this to discussion and have updated it with the resolution. In short, we've discussed the above elsewhere, have already pushed some changes, we were also able to find shortcoming and captured them in issues and todos.