replace engine.Unit with ComponentManager in Pusher Engine #6747
Conversation
Using #4219 as an example. Instead of starting new goroutines or directly processing messages in a blocking way, messages are added to a queue that a worker pulls from. The Pusher engine still currently implements network.Engine rather than network.MessageProcessor.
Codecov ReportAttention: Patch coverage is
Additional details and impacted files@@ Coverage Diff @@
## master #6747 +/- ##
==========================================
- Coverage 41.23% 41.13% -0.11%
==========================================
Files 2060 1896 -164
Lines 182635 169042 -13593
==========================================
- Hits 75316 69533 -5783
+ Misses 101011 93862 -7149
+ Partials 6308 5647 -661
Flags with carried forward coverage won't be shown. Click here to find out more. ☔ View full report in Codecov by Sentry. 🚨 Try these New Features:
|
Because the event processing now happens in a worker, any errors raised within it are no longer visible to the caller of Process(). Because the test checked for error status, moved the tests to the same package and call the internal processing function directly.
When using Unit, calling Ready would also start the engine. With ComponentManager, we additionally need to invoke Start.
| asSCGMsg := asEngineWrapper.Payload.(*messages.SubmitCollectionGuarantee) | ||
| originID := asEngineWrapper.OriginID | ||
|
|
||
| _ = e.process(originID, asSCGMsg) |
There was a problem hiding this comment.
Seems incorrect to fully discard any error, should likely be logged instead?
There was a problem hiding this comment.
Yes, we should be handling all possible error cases. In general, all functions which return an error should document all possible expected error cases in their documentation. Callers which observe an error return that they don't know explicitly how to handle safely, should propagate the error further up the stack. Eventually this results in either the component or entire process restarting.
Here is the part of our coding guidelines talking about error handling: https://github.com/onflow/flow-go/blob/master/CodingConventions.md#error-handling. In short, the rule is to prioritize safety over liveness.
Practically, the way we would do this here is using the irrecoverable.SignalerContext. That has a Throw method which will signal to restart the component or process.
I usually propagate errors all the way up the stack to where the SignalerContext is defined (here, that would be in inboundMessageWorker) then do ctx.Throw(err). You can also use irrecoverable.Throw to throw any context.Context.
Change Suggestions:
- In general, make sure all functions in the file which return an error document their expected error returns (or the lack of expected errors). Here is an example of error documentation.
- Propagate the error value to
inboundMessageWorkerandThrowunexpected errors at that frame.
There was a problem hiding this comment.
agree with your comment that discarding any error is bad. Please take a look at our Coding Conventions, if you haven't already. Tldr:
- just discarding any error: very bad
- logging error and continuing: still bad (this is the pattern of continuing on best effort basis, which is not a viable approach for high-assurance software systems)
- desired:
- explicitly check error type and only continue on errors that are confirmed to be benign.
- all other errors (outside of the specified benign code path) are fatal
The pusher.Engine is a bit of a simpler case, because it only distributes information to other nodes. So swallowing errors does not undermine safety as it would for many other Engines. BUT, we already had cases where nodes were logging OutOfMemory errors for hours, but the code just dropped those and the node was dysfunctional without the software noticing it ... so even if it is not safety critical, we still generally expect that errors are not swallowed (and neither logged and then just discarded).
There are exceptions to this rule, e.g. if it would be a massive time investment to clear up technical debt and thoroughly documenting possible benign errors. But then, a detailed comment is absolutely required why it is always fine to log and discard any errors in that particular case.
I would suggest to leave error handling for a bit and come back to it after some iterations. After having implemented some of the cleanup and simplifications I suggest further below, I suspect that error handling will become a lot more straightforward.
| // Done returns a done channel that is closed once the engine has fully stopped. | ||
| func (e *Engine) Done() <-chan struct{} { | ||
| return e.unit.Done() | ||
| func (e *Engine) processInboundMessages(ctx context.Context) { |
There was a problem hiding this comment.
Should probably make sure functions have doc comments
There was a problem hiding this comment.
Yes, please add a short godoc comment to each function (especially public functions, but ideally private ones too).
|
Issue for the present PR: https://github.com/onflow/flow-go-internal/issues/7018 (it helps reviewers when its linked in the PR description, so they can read up on the context) |
There was a problem hiding this comment.
The overall structure of changes looks good. I've added some thoughts related to transitioning to network.MessageProcessor, which we can discuss separately.
In terms of things to address in this PR:
- Make sure all functions have minimal documentation, including a description of expected error returns (if the function returns any
error). - Make sure invocations of a function that returns expected errors explicitly handles those errors.
- Throw unexpected top-level errors to the
SignalerContext
| asSCGMsg := asEngineWrapper.Payload.(*messages.SubmitCollectionGuarantee) | ||
| originID := asEngineWrapper.OriginID | ||
|
|
||
| _ = e.process(originID, asSCGMsg) |
There was a problem hiding this comment.
Yes, we should be handling all possible error cases. In general, all functions which return an error should document all possible expected error cases in their documentation. Callers which observe an error return that they don't know explicitly how to handle safely, should propagate the error further up the stack. Eventually this results in either the component or entire process restarting.
Here is the part of our coding guidelines talking about error handling: https://github.com/onflow/flow-go/blob/master/CodingConventions.md#error-handling. In short, the rule is to prioritize safety over liveness.
Practically, the way we would do this here is using the irrecoverable.SignalerContext. That has a Throw method which will signal to restart the component or process.
I usually propagate errors all the way up the stack to where the SignalerContext is defined (here, that would be in inboundMessageWorker) then do ctx.Throw(err). You can also use irrecoverable.Throw to throw any context.Context.
Change Suggestions:
- In general, make sure all functions in the file which return an error document their expected error returns (or the lack of expected errors). Here is an example of error documentation.
- Propagate the error value to
inboundMessageWorkerandThrowunexpected errors at that frame.
| // TODO length observer metrics | ||
| inbound, err := fifoqueue.NewFifoQueue(1000) | ||
| if err != nil { | ||
| return nil, fmt.Errorf("could not create inbound fifoqueue: %w", err) | ||
| } |
There was a problem hiding this comment.
Let's add those length metrics now for completeness. You can use this engine as an example. We'll also need to add a parameter for the mempoolMetrics to the constructor.
| // Done returns a done channel that is closed once the engine has fully stopped. | ||
| func (e *Engine) Done() <-chan struct{} { | ||
| return e.unit.Done() | ||
| func (e *Engine) processInboundMessages(ctx context.Context) { |
There was a problem hiding this comment.
Yes, please add a short godoc comment to each function (especially public functions, but ideally private ones too).
| // TODO convert to network.MessageProcessor | ||
| var _ network.Engine = (*Engine)(nil) | ||
| var _ component.Component = (*Engine)(nil) |
There was a problem hiding this comment.
It may also be a good exercise to make the change to MessageProcessor here, though likely as a separate PR.
I'll add some thoughts below; I suggest we spend some time discussing this live as well. There is no need to implement any of the comments below before we have a chance to discuss.
The pusher engine is unusual in a couple of ways:
- It is using an old framework for internal message-passing, where two components in the same process would pass messages between each other using standard methods (
SubmitLocalandProcessLocal). The meaning of the message was carried by its type (here:messages.SubmitCollectionGuarantee) and the logical structure was similar to that of messages received over the network. Message processing logic needed to potentially handle inputs originating both locally and remotely. - It does not expect to receive any inbound messages from non-local sources. It is send-only from a networking perspective.
Replacing network.Engine with network.MessageProcessor there are a few opportunities for simplifying the implementation:
- Currently the internal message-passing is done through
SubmitLocal. We can removeSubmitLocalandProcessLocaland replace them with a context-specific function (and corresponding interface type, for use in the caller).- A reference example is the
Complianceinterface, which is exposed by the compliance engine for local message-passing. - For this reason,
engine.MessageHandlermay not be needed in this context. It is better suited for cases where we are handling external messages. - We then do not need to check originIDs and do not need both
onSubmitCollectionGuaranteeandSubmitCollectionGuaranteemethods - Now, this context-specific function is the only way for local messages to enter the
pusherengine, which will simplify our network message processing logic 👇
- A reference example is the
- In order to satisfy
network.MessageProcessor, we need aProcessmethod, which accepts messages from remote network sources. Since this engine does not expect any remote messages, we should reject all inputs toProcess.- In general, when we observe an unexpected message from another node, we should flag it as a possible Byzantine message. In the future, these messages would result in a slashing challenge targeting the originator of the message. For now, we can log a warning with the relevant log tag.
There was a problem hiding this comment.
+1. Had a similar thought 👉 my comment
| // started. | ||
| func (e *Engine) Ready() <-chan struct{} { | ||
| return e.unit.Ready() | ||
| func (e *Engine) inboundMessageWorker(ctx irrecoverable.SignalerContext, ready component.ReadyFunc) { |
There was a problem hiding this comment.
Please add a short godoc for this method
| }) | ||
| err := e.messageHandler.Process(e.me.NodeID(), event) | ||
| if err != nil { | ||
| engine.LogError(e.log, err) |
There was a problem hiding this comment.
| engine.LogError(e.log, err) | |
| // TODO: do not swallow errors here, remove this function when transitioning to `network.MessageProcessor` | |
| engine.LogError(e.log, err) |
| }) | ||
| err := e.messageHandler.Process(originID, event) | ||
| if err != nil { | ||
| engine.LogError(e.log, err) |
There was a problem hiding this comment.
| engine.LogError(e.log, err) | |
| // TODO: do not swallow errors here, remove this function when transitioning to `network.MessageProcessor` | |
| engine.LogError(e.log, err) |
| err := e.messageHandler.Process(originID, event) | ||
| if err != nil { | ||
| engine.LogError(e.log, err) | ||
| } | ||
| } | ||
|
|
||
| // ProcessLocal processes an event originating on the local node. |
There was a problem hiding this comment.
Currently we are directly returning the result of MessageHandler.Process. That function documents a possible expected error return type. At this point we should either:
- include that expected error type as a possible return value from this function
- explicitly handle the expected error, or explain why it is not actually expected in this specific context
In this case, the event is originating from a local source (another component in the same process). In general we trust locally generated inputs, so here we could add a comment explaining that, in this context, an IncompatibleInputTypeError should be treated as an exception.
Here is an example where we observe a sentinel error from function "f" and document why, in the context of the caller of "f", that error is an exception.
There was a problem hiding this comment.
Thanks for the PR. My comments are not a full review (yet). Rather I am aiming at highlighting the areas to pay attention to. When working on an issue, part of the goal is to address the issue, though fixing the issue is frequently not sufficient: one of the major challenges with implementing a high-assurance software system such as Flow is managing intellectual complexity, code complexity and component isolation. When an engineer is required to grok the entire code base or a large portion thereof in order to make any change, then we have lost the ability to extend and maintain the code base on business-relevant time scales.
The Pusher Engine is from a very early time when most engineers knew most of the code. As the code base grows more complex, by adding functionality towards the mature architecture, we need more precision and rigour in the implementation.
We appreciate that it is super hard for new engineers to identify technical debt as such. Therefore, I have provided detailed context why certain aspects are technical debt. As you grow more experienced and familiar with our best practises, it will become easier to spot tech debt. Nevertheless, I want to be clear about expectations: cleaning up tech debt in/around the code you touch as time permits. Always leave the code in a better state than you found it, especially wrt:
- documentation in general (fixing outdated goDoc and adding missing docs)
- error documentation in particular and error handling
- code clarity and removing unnecessary abstraction layers
- clearly apparent reasoning for correctness (either by the code itself or through added documentation)
- API safety (is it hard to use the API incorrectly?)
- test coverage
For quite some time, we will be helping you with spotting areas of tech debt and how to address them. If you see something that looks like tech debt and you don't know how to fix is, please ask - I loved your comment here. I can give you one recommendation: if you don't see anything to be improved at a piece of code (aside from implementing your issue) you should probably ask somebody for input. The code we touch tends to generally be older code (because the newer code often has already the desired features) and the older code is generally in dire need for cleaning up tech debt. Its frequent that engineers spend just as much time, frequently even longer, cleaning up tech debt than working on the actual issue. I think this will also be the case for the Pusher Engine.
| // Engine is the collection pusher engine, which provides access to resources | ||
| // held by the collection node. | ||
| type Engine struct { |
There was a problem hiding this comment.
very very ambiguous phrasing: "provides access to resources held by the collection node." By which means does it "provides access" (request-response?). What specific "resources" does the engine provide access to (all of the resources, many, ...?).
| // Engine is the collection pusher engine, which provides access to resources | |
| // held by the collection node. | |
| type Engine struct { | |
| // Engine is part of the Collection Nodes. It broadcasts finalized collections | |
| // that the cluster generates to the broader network. | |
| type Engine struct { |
| asSCGMsg := asEngineWrapper.Payload.(*messages.SubmitCollectionGuarantee) | ||
| originID := asEngineWrapper.OriginID | ||
|
|
||
| _ = e.process(originID, asSCGMsg) |
There was a problem hiding this comment.
agree with your comment that discarding any error is bad. Please take a look at our Coding Conventions, if you haven't already. Tldr:
- just discarding any error: very bad
- logging error and continuing: still bad (this is the pattern of continuing on best effort basis, which is not a viable approach for high-assurance software systems)
- desired:
- explicitly check error type and only continue on errors that are confirmed to be benign.
- all other errors (outside of the specified benign code path) are fatal
The pusher.Engine is a bit of a simpler case, because it only distributes information to other nodes. So swallowing errors does not undermine safety as it would for many other Engines. BUT, we already had cases where nodes were logging OutOfMemory errors for hours, but the code just dropped those and the node was dysfunctional without the software noticing it ... so even if it is not safety critical, we still generally expect that errors are not swallowed (and neither logged and then just discarded).
There are exceptions to this rule, e.g. if it would be a massive time investment to clear up technical debt and thoroughly documenting possible benign errors. But then, a detailed comment is absolutely required why it is always fine to log and discard any errors in that particular case.
I would suggest to leave error handling for a bit and come back to it after some iterations. After having implemented some of the cleanup and simplifications I suggest further below, I suspect that error handling will become a lot more straightforward.
| messageHandler := engine.NewMessageHandler(log, notifier, engine.Pattern{ | ||
| Match: engine.MatchType[*messages.SubmitCollectionGuarantee], | ||
| Store: &engine.FifoMessageStore{FifoQueue: inbound}, | ||
| }) |
There was a problem hiding this comment.
I would recommend to not use the MessageHandler here for the following reasons:
- I think it is good to apply the patter of the
MessageHandlerat least once by hand ("learning by doing") - The
MessageHandleradds a layer of abstraction, if we can do without it that would be preferable in my opinion. - In this particular case, the Engine only processes a single input type
SubmitCollectionGuarantee. We have tried it before and the message handler is actually more code compared to just doing the same logic by hand.
There was a problem hiding this comment.
Please take a look at the core processing logic in the Engine and lets try to remove layers of complexity and indirection where possible:
flow-go/engine/collection/pusher/engine.go
Lines 182 to 191 in b5418dd
Looking at this code raises a few questions:
-
we are only propagating collections that are originating from this node (👉 code). But the exported method
SubmitCollectionGuaranteeis not checking this? Why not?- If it is an important check, we should always check it. If we are fine with collections from other nodes being broadcast, then we don't need the check in
onSubmitCollectionGuaranteeeither.. - turns out,
SubmitCollectionGuaranteeis only ever called by thepusher.Engineitself (not even the tests call it).
Lets simplify this: merge
SubmitCollectionGuaranteeandonSubmitCollectionGuarantee. - If it is an important check, we should always check it. If we are fine with collections from other nodes being broadcast, then we don't need the check in
-
Further digging into the code, I found that the
originIDis just completely useless in this context:- the
pusher.Engineis fed only from internal components (sooriginIDis always filled with the node's own identifier) - the
pusher.Enginedrops messages that are not from itself. Just hypothetically, lets assume that there were collections withoriginIDother than the node's own ID. ❗Note that those messages would be queued first, wasting resources and then we have a different thread discard the Collection after dequeueing. Checks like that should always be applied before queueing irrelevant messages.
What you see here is a pattern we struggle with a lot: the code is very ambiguous and allows a whole bunch of stuff that is never intended to happen in practise. This is an unsuitable design for engineering complex systems such as Flow, as we then have to manually handle undesired cases.
- Example 1: messages other than SubmitCollectionGuarantee. In comparison, if we had a method that only accepted
SubmitCollectionGuarantee, the compiler would enforce this constraint for us as opposed to us having to write code. Often, people don't bother to manually disallow undesired inputs, resulting in software that is doing something incorrect in case of bugs as opposed to crashing (or the code not compiling in the first place) - Example 2:
originID. We employ a generic patter which obfuscates that we don't need anoriginIDin this particular example, because theEngineinterface does not differentiate between node-external inputs (can be malicious) vs inputs from other components within its own node (always trusted) and mixes modes of processing (synchronous vs asynchronous).
For those reasons, we deprecated the
Engineinterface. Lets try to update thepusherand remove the engine interface (for concrete steps towards this, please see my next comment). - the
For Flow, the APIs should generally be as descriptive and restrictive as possible and make it hard to feed in wrong / mismatching inputs. Writing extra lines of boilerplate code is not a challenge for Flow. In contrast, it is challenging if you have to dig through hundreds or thousands of lines of code because the APIs are too generic and you need to know implementation details to use them correctly.
| // SubmitLocal submits an event originating on the local node. | ||
| func (e *Engine) SubmitLocal(event interface{}) { |
There was a problem hiding this comment.
As an exercise, try to figure out which other components are feeing inputs into the pusher.Engine. We are lucky here because we have a rarely used message type SubmitCollectionGuarantee. But just imagine you would see a log that says no matching processor for message of type flow.Header from origin. With the Engine interface it is way too time intensive to figure out which components inside a node are feeding which other components with data.
Turns out that the Finalizer is the only component that is feeding the pusher.Engine (other than messages arriving from the networking layer). Let's make our lives easier in the future by improving the code:
- introduce a dedicated interface
GuaranteedCollectionPublisher. It only has one type-safe methodthis method just takes the message and puts it into the queue. Then we know that the queue will only ever contain elements of typeSubmitCollectionGuarantee(guarantee *flow.CollectionGuarantee)
flow.CollectionGuarantee. (currently, the Finalizer creates the wrappermessages.SubmitCollectionGuaranteearoundCollectionGuarantee... and thepusher.Engineunwraps it and throws the wrapper away 😑 ). - change the Finalizer and all layers in between to work with the
GuaranteedCollectionPublisherinterface. This improves code clarity: currently, engineers would need to know what functionality was backing the genericEngineinterface, e.g. here: because the right type of inputs need to be fed into thepusher.Engine. When using theGuaranteedCollectionPublisherinterface we have the following benefits:- the type itself carries information about what functionality is behind this interface. Huge improvement of clarity.
- its easy to find all places that can feed the
Pusherwith data, because we work with concrete interfaces and context-specific methods (e.g.SubmitCollectionGuarantee). - the compiler will reject code that feeds inputs with incompatible type into the
Pusher
- In my opinion, it would improve clarity if we were to discard all inputs from the
network.MessageProcessor- conceptually they are messages from different node, which they are already broadcasting. ThePusheris for broadcasting our own messages. But at the moment, inputs from node-internal components as well as messages from other nodes passing through the networking layer share the same code path in thepusher.Enging, which makes it non-trivial to figure out where messages are coming from.
To summarize: one of the main benefits of the Component interface is that it provides lifecycles methods, but no methods for ingesting data - on purpose. This is because all data from components within the node should be passed via context-specific methods. The only exception is the interface network.MessageProcessor, which would still call Process with a generic message type.
Sorry for the lengthy comment. I hope you see how the very ambiguous structure of the Engine makes it really hard to determine from the code what we are expecting and where to drop messages, because its all hidden behind layers of ambiguous interfaces in a huge code base.
| func New(log zerolog.Logger, net network.EngineRegistry, state protocol.State, engMetrics module.EngineMetrics, colMetrics module.CollectionMetrics, me module.Local, collections storage.Collections, transactions storage.Transactions) (*Engine, error) { | ||
| // TODO length observer metrics | ||
| inbound, err := fifoqueue.NewFifoQueue(1000) |
There was a problem hiding this comment.
I am not sure about the queue capacity we should pick here. I think we should have a good argument (and document it!) of why the number we chose makes sense. If we don't want to go through the trouble of defining a sensible limit, I think we could make this clear by using an unbounded queue (implementation).
Anyway, I think a limit makes sense:
- In a nutshell, a collection must reference a block and expires
$\texttt{DefaultTransactionExpiry}$ = 600 blocks thereafter(its a subtle mechanic, feel free to ignore)flow-go/model/flow/constants.go
Lines 18 to 22 in a164070
- consensus nodes produce about
$\rho$ = 1.25 blocks per second - collector clusters produce up to
$\varrho$ = 3 collections per second
So very very roughly, between a collection being first proposed (usually referencing the newest known block) to the collection expiring, the collector cluster will produce
The downside is that collections might sit in the queue for quite some time, if several hundreds are before them. That could be an argument of queuing only much fewer collections (e.g. 200), so collection that don't make it out fast are dropped.
Another argument we could make is actually using a LIFO queue. If the queue is nearly empty, there is little difference. However, if there is a large backlog of collections to be broadcast, I think its a valid argument of broadcasting the newest first, at least then, some collections make it in time as opposed to all of them being queued past their expiry. Just some thoughts.
@jordanschalm what do you think?
Using #4219 as an example. Instead of starting new goroutines or directly processing messages in a blocking way, add messages to a queue that a worker pulls from.
The Pusher engine still currently implements
network.Enginerather thannetwork.MessageProcessor.