[DataAvailability] Add results forest workqueue manager by peterargue · Pull Request #7537 · onflow/flow-go

peterargue · 2025-06-23T20:47:54Z

Closes: #7465

This PR builds on #7534, and implements the worker queue manager for execution ExecutionResult processing pipelines.

peterargue · 2025-06-24T20:56:44Z

+		return BlockStatusCertified, fm.config.RequiredBlockStatus == BlockStatusCertified
+	}
+
+	if height > fm.sealedHeight {


this is not sufficient because there could be results for unfinalized blocks in the forest, which could have conflicting heights.

Instead, this also needs to lookup the block by height and ensure that the returned block ID matches the results' block ID. If they match, then the block is definitely finalized, and this height check is then OK.

the forest already consumes and tracks finalization. Maybe we can make the forest update the containers along the finalized fork to persist their status.

⚠️ The reason why I am saying this is because your are planning to mixg information from different sources here -- which is always a risk for inconsistent information. You are planning to use the global protocol state, instead of the eventually-consistent state the Forest maintains. At the moment, it is clear. But when the code evolves, mixing information from different sources is easily overlooked by just assuming that its the same state. But the Forest is not the same state as the global protocol state -- its a subset with less information, and things that are true for the global protocol state (like knowing all the ancestors) might not be true for the Forest.

https://www.notion.so/flowfoundation/Reactive-State-Management-in-Distributed-Systems-A-Guide-to-Eventually-Consistent-Designs-22d1aee1232480b2ad05e95a7c48a32d?source=copy_link

My recommendation would be:

differentiate:

auxiliary information that is immutable by protocol convention (like looking up the executed block header for an execution result)

and deductions from the state (like connectedness, finality, sealing).

is always going to be the same, no matter which component queries that information. 2. changes over time. So if you query different sources, like the Forest vs the Global State, one might know aspects that are still unknown to the other.

This has the potential to get super complicated really quickly. You need to be super disciplined and only with a single source of truth. And your source of truth here is the Forest. Its more work in the beginning, but I am pretty convinced it is worth in the long run:

incorporate all deductions (2.) into the forest first, and create back-fill mechanisms. For finality that is pretty straight forward: the backfill process first adds all blocks that might have been missed, and after each block, it may update the finality status (caution use information driven approach: "at least up to this block, finality has progressed" -- not "this is the latest finalized block").

Similarly for sealing.

And then only work with information that is in the forest. A block might be finalized or sealed but the Forest does not yet know. That is fine, the backfill process will guarantee that the forest learns this information eventually. And when the forest learns this, we trigger the respective actions.

In any case avoid triggering actions based on the state of a different component. Because then, you have to argue why, despite the information gap between the forest and the global state, we still do always the correct thing. You remember this subtly now, but in 6-12 months after having added a few performance optimizations driving up complexity, I am very skeptical that anyone will remember when touching this specific line that a block could be finalized in global state but not yet in the forest.

…-manager

AlexHentschel

first batch of comments ... still have some code to cover.

AlexHentschel · 2025-07-11T04:17:01Z

+// OnBlockStatusUpdated signals that the block status has been updated.
 // It finds all vertices for results of blocks that conflict with the finalized block and abort them.
-func (rf *ResultsForest) OnBlockFinalized(finalizedBlockID flow.Identifier, parentBlockResultIDs []flow.Identifier) {
+func (rf *ResultsForest) OnBlockStatusUpdated(finalized *flow.Header, sealed *flow.Header, parentBlockResultIDs []flow.Identifier) error {


It seems we are generalizing this function to not only consume finalization events, but status updates in more general (?) Though, the method implementation and documentation is still strongly focused on "finalization" (?) I would suggest to use the most specific name.

AlexHentschel · 2025-07-11T04:33:57Z

 type ResultsForest struct {
 	log                         zerolog.Logger
 	forest                      forest.LevelledForest
+	manager                     *ForestManager


somewhat related to my prior comment:

We are embedding the ForestManager here. This would mean that the Forest would need to do something to the ForestManager. My intuitive approach would be the reverse: the manager is doing stuff with the Forest. In other words, the manger knows the forest that it is managing, but not vice versa.

I think the prior name OnBlockFinalized of the method that is now called OnBlockStatusUpdated is a good indicator that reversing the dependency might be useful. Because the Forest seems to only care about finalization, so it can orphan stuff.

If the manager needs block events more generally, I would opt for either of the following approaches:

maybe the ForestManager can ingest the event and pass it onto the forest if the event is of type OnBlockFinalized

alternatively, I think its totally fine for the ingestion engine to receive the events and then forward them to either or both the manager and the forest.

In my head, the forest is a fork-aware data storage component. By virtue of being fork-aware it is natural for the forest to understand concepts like pruning and orphaning (via finalization and sealing).
Though, in my head, the higher-level business logic should live outside of the forest and operate on the forest. In other words, the forest provides the low-level primitives to query and operate on the graph structure the forest maintains. But the forest could hopefully be largely oblivious what the higher-level business logic is actually doing. To me, the ForestManager seems to be more part of the higher-level business logic (because it's concerned with what is happening inside the containers).

It's probably too premature for me to suggest changing this now. We could revisit this point when you implement the ingestion engine, and then invert the dependencies.

AlexHentschel · 2025-07-11T04:45:29Z

+// OnBlockStatusUpdated is called when a new block is finalized.
+// This is used to notify the forest that some results may now be processable.
+func (fm *ForestManager) OnBlockStatusUpdated(finalized, sealed *flow.Header) {
+	fm.mu.Lock()
+	defer fm.mu.Unlock()
+
+	fm.sealedHeight = sealed.Height
+	fm.finalizedHeight = finalized.Height
+
+	fm.notifier.Notify()
+}


I would recommend to think about this entire implementation as an information driven system. Example:

you know that blocks ... 8, 9, 10 are finalized.

now I come and tell you "I just learned that block 9 is finalized"

if I asked you, what's the latest finalized block, I would intuitively expect that you answer 10

your implementation in contrast will forget that block 10 is already finalized and answer 9.

your implementation makes strong assumptions about in-order delivery of finalization events ... that's fine if doing it otherwise incudes a lot of complexity, but if you are expecting monotonically increasing finalization for correctness, you would ideally error on inputs that violate this assumption (if easily possible)

either way, you should be checking that height never decreases

either error if that is violated

or no-op in the spirt of building an information-driven system (my personal preference)

AlexHentschel · 2025-07-11T04:53:53Z

 	rf.mu.Lock()
 	defer rf.mu.Unlock()

 	// 1. Get all ExecutionResults for the finalized block's parent (done by caller)


⚠️ To the best of my understanding, this logic works only correctly, if all parent block results that exist somewhere in the network are known. I don't think that is tractable.

I just noticed this. So the forest is the component that maintains the graph. But the caller tells the forest what the parent results are? Isn't that what we have the forest for, to know the parent and child relationships?

I think we can conceptually decompose this into two sub-problems:

the forest already knows the parent's result's. In this case, the forest can find the parent independently and doesn't need the parent result be handed in externally.

the forest does not yet know the parent's results. In this case, we know some forks are already abandoned.

But there is no risk of accidentally processing those abandoned forks for now, because there is still a parent missing.

Once we add a result, we can check if it conflicts with the finalized block. Only then we can orphan the forks.

I think we need 1. and 2.ii. To the best of my understanding, your solution doesn't cover 2.ii, so we would need to add it anyway. And once we have 1. and 2.ii, we don't need to rely on external results anymore, because the forest knows everything to decide whether forks are abandoned.

I think generally, this is a very important encapsulation point. We want to avoid the forest to rely on external information, because (first) that introduces subtle requirements for external code that in tendency are easily overlooked and accidentally broken in my opinion and (second) we pretend that some external logic could magically know all parent block results that exist somewhere in the network, which also is not the case.

I am wondering if we are on a deep level repeatedly bumping into the same pattern:

I notice the pattern, where you have your external business logic store some insights into the tree, when that external business logic gains that new insight.

But fundamentally, the forest has an incomplete view of the block and execution result tree. There is an information gap. Maybe it helps for me to quickly take a tangent to logic and information theory. In those domains it is frequently the case that some statement is true (e.g. this algorithm always eventually terminates) that cannot be confirmed by an algorithm to be true (e.g. the halting problem).

Similarly, we have statement here like "this fork is orphaned", which the result forest cannot confirm to be true when it is missing ancestors.

I think your solution to this problem is just "I tell the forest that X is true". However, lets remember that the forest has an incomplete view of all the forks. Even though the consensus follower might know that a fork is orphaned, the forest might not have enough coverage of the forks to even persist that information. Blocks and results might be added later, and by the time they are added, the events from the consensus follower have long passed.

Where I feel your algorithms frequently break is when the forest doesn't know stuff because it might have rejected information earlier, which just hasn't been backfilled yet.

The solution is surprisingly always the same: formulate your algorithms such that they decide upon successful insertion into the forest, whether the result satisfies a property of interest (orphaned, ready for processing etc). Its embracing the eventually-consistent design.

AlexHentschel · 2025-07-11T04:54:23Z

 	// 2. For each of these results, get all child vertices
 	// 3. For each child vertex, cancel if it does not reference the finalized block

+	finalizedBlockID := finalized.ID()


this should be computed before acquiring the lock

AlexHentschel · 2025-07-11T06:09:52Z

+// String returns the string representation of the block state
+func (bs BlockStatus) String() string {
+	switch bs {
+	case BlockStatusSealed:
+		return "sealed"
+	case BlockStatusFinalized:
+		return "finalized"
+	case BlockStatusCertified:
+		return "certified"
+	default:
+		return "unknown"
+	}
+}


In my head, it would be natural to store this status with the execution results -- because in this context, we use it mostly as a descriptor for the result than the block.

that's a good point, and I think this would simplify some of the tracking in the result forest as well.

AlexHentschel · 2025-07-11T06:20:22Z

+		return BlockStatusFinalized, fm.config.RequiredBlockStatus == BlockStatusCertified || fm.config.RequiredBlockStatus == BlockStatusFinalized
+	}
+
+	return BlockStatusSealed, true


I don't understand why we can categorically label blocks below the sealed height as sealed. What is with blocks that are orphaned. Similarly, are all blocks below or equal the finalized height also finalized - what is with orphaned blocks.

This might be an instance of the general pattern which I tried to discourage in my earlier comment:

what the status of the executed block for a result is should be updated on insertion into the tree and thereby a property derived from the tree.

here, again I think its something we are telling the tree. And I don't understand why that always yields the correct result.

For example, you write: "the results forest will abandon any results that conflict with finalized or sealed results."

I would argue that this statement does not hold in its generality (my comment here). The forest abandons forks that it knows about upon finalizing a block. But there could be additional forks added later, which we couldn't mark as abandoned while finalizing because they where simply not yet known. Now they are known, but the finalization event has passed.

I think these block status values need to be determined on insertion into the tree, based on the information that is already stored in the tree. I am relatively certain that this is conceptually always possible, might not be easy see right away how to implement it, but algorithmically it should be possible.

makes sense. I'll make this change in the results forest PR and link the commit here.

I adjusted in ce06fb2

…-manager

AlexHentschel

This is my last batch of comments on this PR.
A decent portion of my comments touch on aspects that I tried to explain more broadly in Reactive State Management in Distributed Systems: A Guide to Eventually Consistent Designs. Hope it makes sense.

If some of my comments are too brief or convoluted, I would suggest to ask Yurii. Either way, don't feel blocked if you don't know how to address some of my comments while I am away on vacation. My gut feeling is that we could also change all aspects I have commented on with a moderate time investment later.

AlexHentschel · 2025-07-15T05:33:22Z

+	// RequiredExecutors is the set of executor IDs that must have provided receipts
+	// At least one of these executors must have produced a receipt for the result to be processed
+	RequiredExecutors map[flow.Identifier]struct{}


Probably beyond the scope of this PR, and easy enough to refactor when changing it later: maybe we can generalize this to a function taking the set of ENs that have produced a receipt and returning a bool value.

flow-go/engine/access/ingestion2/execution_result_container.go

Line 24 in 8d6b396

receipts map[flow.Identifier]*flow.ExecutionReceiptMeta // map from ExecutionReceipt.ID -> ExecutionReceiptMeta

AlexHentschel · 2025-07-15T18:29:10Z

+	}, nil
+}
+
+// WorkerLoop is a component.ComponentWorker that adds processable containers to the work queue.


Suggested change

// WorkerLoop is a component.ComponentWorker that adds processable containers to the work queue.

// WorkerLoop is a [component.ComponentWorker] that adds processable containers to the work queue.

AlexHentschel · 2025-07-15T18:48:15Z

+		workQueue:     workQueue,
+		config:        config,
+		notifier:      notifier,
+		maxQueueSize:  maxQueueSize,


I would suggest to revisit the naming here. When you write maxQueueSize and further above you have a workQueue, that implies to me that workQueue cannot hold more than maxQueueSize items. I am not sure if that's actually the case that workQueue has a size restriction, but if it does the size should be a property of the work queue and not passed around separately.

Though, I assume that there is no hard limit for the workQueue's size and maxQueueSize is rather an indicator of how many work items we desire to be minimally in the work queue (in scenarios where there is a backlog of work). So if my understanding is correct maxQueueSize is not a hard upper limit (what the current name is suggesting) but rather a desired minimal number of queued up tasks to not leave workers idling. Hence, I would suggest a wording like desiredQueuedTasks including a revision of the goDoc here:

flow-go/engine/access/ingestion2/forest_manager.go

Lines 101 to 102 in 5d344c9

// maxQueueSize is the maximum size of the work queue

maxQueueSize uint

AlexHentschel · 2025-07-15T20:21:07Z

+	latestPersistedSealedResultID, _ := fm.resultsForest.latestPersistedSealedResult.Latest()
+	latestPersistedSealedResultContainer, found := fm.resultsForest.GetContainer(latestPersistedSealedResultID)
+	if !found {


⚠️ I think this is prone to subtle race conditions:

you ask: what is the latest sealed result container and get an ID

then pruning advances, and your container is removed

then you ask for the container

I think you cannot conclude that the Forest is in an inconsistent state, because your retrieval is not atomic.

AlexHentschel · 2025-07-15T20:27:07Z

+		latestPersistedSealedResultContainer, err := fm.getLatestPersistedSealedResultContainer()
+		if err != nil {
+			ctx.Throw(err)
+			return
+		}
+
+		fm.traverseDescendants(latestPersistedSealedResultContainer)


As we have discussed in person, I think this traversal is not necessary:

the forest can exclusively decide on addition

If a result is inserted or its state is changed, we check whether the result is processable

if it now is processable, we try to add all descendants (I would drop the maxQueueSize and just add everything to the queue that we can) in a breadth-first search way.

if a result is not processable, either (a) its state will change later at which point we will re-evaluate its processability or (b) if the overall protocol abandons the result, it may never receive a state update and be eventually pruned.

To be precise, I think this entire method can be removed. I would suggest to remove it, simply because we want to embrace the eventually-consistent design paradigm with deduction upon insertion. For one as a learning opportunity, but even more importantly because mixing iterative and event-driven designs like here often results in major complexity, maintenance, and concurrency (performance) obstacles in the long run.

AlexHentschel · 2025-07-15T20:33:25Z

+	fm.sealedHeight = sealed.Height
+	fm.finalizedHeight = finalized.Height


would suggest to move this into the forest. The manager can query the forest.

AlexHentschel · 2025-07-15T20:40:23Z

+		select {
+		case <-ctx.Done():
+			return
+		case <-notifierChan:


I like your approach of the Manager having its own worker routines (it could probably even have more than one, but I don't think that would be speeding up things).

I would suggest:

OnReceiptAdded(..) and OnBlockStatusUpdated put notifications into a queue inside the manager. And then call Notify as you do here:

flow-go/engine/access/ingestion2/forest_manager.go

Line 218 in 5d344c9

}

My gut feeling is that it would be sufficient to just add the respective ExecutionResultContainer in a queue. Essentially we are saying: "something has changed about this container, please re-evaluate processability of the container and all its descendants"

why the queue. Well for once because it makes things more obvious: you could attempt to process a notification for a container that has already been pruned. The queue makes this very obvious. Furthermore, it provides a clear resourcing separation between threads serving the forest and the threads serving the worker (if we want to).

AlexHentschel · 2025-07-15T20:42:50Z

Wondering if a different name could be beneficial: I feel this component manages the work resulting from the graph stored in the forest. But it does not manage the graph/ ResultForest itself. Hence, I would rename it to ResultsProcessingManager

…-manager

Co-authored-by: Alexander Hentschel <alex.hentschel@flowfoundation.org>

…-manager

[DataAvailability] Add results forest workqueue manager

23c3f9c

peterargue requested a review from a team as a code owner June 23, 2025 20:47

remove old todos

a2c87d2

peterargue commented Jun 24, 2025

View reviewed changes

peterargue added 2 commits June 27, 2025 10:11

Merge branch 'peter/results-forest-workers' into peter/results-forest…

02de7b1

…-manager

cleanup godocs and improve logic

5d344c9

AlexHentschel reviewed Jul 11, 2025

View reviewed changes

Merge branch 'peter/results-forest-workers' into peter/results-forest…

b3f649e

…-manager

peterargue force-pushed the peter/results-forest-manager branch from 9c5c4de to b3f649e Compare July 11, 2025 19:33

fix merge conflict

8d6b396

AlexHentschel approved these changes Jul 15, 2025

View reviewed changes

peterargue and others added 8 commits August 27, 2025 12:10

Merge branch 'peter/results-forest-workers' into peter/results-forest…

4d1b944

…-manager

Apply suggestions from code review

bcb916f

Co-authored-by: Alexander Hentschel <alex.hentschel@flowfoundation.org>

Merge branch 'peter/results-forest-workers' into peter/results-forest…

a81a289

…-manager

Merge branch 'peter/results-forest-workers' into peter/results-forest…

726cad4

…-manager

rework from code review

f2083c9

Merge branch 'peter/results-forest-workers' into peter/results-forest…

8b8bf9f

…-manager

Merge branch 'peter/results-forest-workers' into peter/results-forest…

dae60a4

…-manager

improve comments

e15acfa

peterargue mentioned this pull request Sep 10, 2025

[DataAvailability] Add results forest and execution result container #7523

Closed

github-actions Bot added the Stale Label used when marking an issue stale. label Oct 30, 2025

peterargue added Preserve Stale Bot repellent and removed Stale Label used when marking an issue stale. labels Nov 5, 2025

	// WorkerLoop is a component.ComponentWorker that adds processable containers to the work queue.
	// WorkerLoop is a [component.ComponentWorker] that adds processable containers to the work queue.

	// maxQueueSize is the maximum size of the work queue
	maxQueueSize uint

		fm.sealedHeight = sealed.Height
		fm.finalizedHeight = finalized.Height

Conversation

peterargue commented Jun 23, 2025 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

AlexHentschel left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

AlexHentschel left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

This branch has not been deployed

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

3 participants

peterargue commented Jun 23, 2025 •

edited

Loading