Update stage_finish.go : notifications to rpc daemon

[AndreaLanfranchi]

Only notify for canonical header obtained from last Finish.

Imho this should be enough. But review !!

[AskAlexSharov]

It's incompatible with docs: https://geth.ethereum.org/docs/rpc/pubsub

[AndreaLanfranchi]

It's incompatible with docs: https://geth.ethereum.org/docs/rpc/pubsub

I'm not arguing but willing to understand. From the specs

notifications are sent for current events and not for past events. If your use case requires you not to miss any notifications than subscriptions are probably not the best option.

Due to Erigon's stage structure is inevitable, particularly on stage cycle 0, all notifications of headers before the very last (the one recorded in Finish) are "past" events.

notifications are stored in an internal buffer and sent from this buffer to the client. If the client is unable to keep up and the number of buffered notifications reaches a limit (currently 10k) the connection is closed. Keep in mind that subscribing to some events can cause a flood of notifications, e.g. listening for all logs/blocks when the node starts to synchronize

During first 2 or 3 cycles of a sync from scratch we might well send out more than 10k notifications.

Maybe we can send out notifications for all processed headers only if notifyTo - notifyFrom < 1024 ? (i.e. the threshold for single transaction cycles ?

Waiting for your comments

[AskAlexSharov]

“send out notifications for all processed headers only if notifyTo - notifyFrom < 1024” i like this logic.

[AndreaLanfranchi]

@AskAlexSharov question ... do we really need to pass NotifyHeaders this unwindTo *uint64 ?
I mean ... an unwind should have already persisted the Finish value after the cycle (before notification) so basically
stages.GetStageProgress(tx, stages.Finish) is the new finishStageAfterSync and we simply need to compute delta amongst
finishStageBeforeSync and finishStageAfterSync
Am I missing something ?

[AndreaLanfranchi]

Also ... context.Context is not used in the function.
(sorry to bother but in C++ we got warnings or errors for unused formal parameters)

[AskAlexSharov]

• Agree, this method was inside of stage_finish in the past. I have an idea:
  something like

		tx.ForEach(kv.Headers, bigEndian(notifyFrom), func(k, headerRLP []byte) error {
			notifier.OnNewHeader(headerRLP)
			return nil
		})

It will do zero-copy broadcast (zero marshal/unmarshal) from db to network of canonical and non-canonical headers from notifyFrom to the end of kv.Headers table in DB.
Then we don't need unwindTo parameter, don't need get stageFinish progress, etc...
It require change signature of ChainEventNotifier.OnNewHeader but it's easy because only one consumer of this API.

• context is used to stop on Ctrl+C - need add it to loop:

if err := libcommon.Stopped(ctx.Done()); err != nil {
			return libcommon.ErrStopped
		}

[AndreaLanfranchi]

of canonical and non-canonical headers

Not sure I understand this. Doesn't RPC Daemon need to be notified of highest canonical header(s) only ?

[AskAlexSharov]

Docs says https://geth.ethereum.org/docs/rpc/pubsub:

newHeads

Fires a notification each time a new header is appended to the chain, including chain reorganizations. Users can use the bloom filter to determine if the block contains logs that are interested to them.

In case of a chain reorganization the subscription will emit all new headers for the new chain. Therefore the subscription can emit multiple headers on the same height.

[AskAlexSharov]

I still think having 1024 as limit does make sense.

[AndreaLanfranchi]

I still think having 1024 as limit does make sense.

1024 or 8096 like canRunCycleInOneTransaction here ? If the latter we can pass the boolean and avoid redoing the math

[AskAlexSharov]

I'm ok with re-using canRunCycleInOneTransaction

[AndreaLanfranchi]

I did my best (my GO skills are at "Hello World" level)

• All headers are notified (canonical or not) within a max span of 1024
• NotifyHeaders now receives the stage "Finish" height after the cycle (which is already computed in StageLoopStep)
• I wish I could print the number of the highest header notified but apparently I am not able to set a capture variable in the walk function of ForEach

I did not dare to change the signature of OnNewHeader as I see there are multiple dependencies which might get broken

Please take a look ... and don't be frightened to be rude :D

[AskAlexSharov]

Thank you!

• About capture variable: declare variable before foreach by var lastKey []byte and inside walk func assign to this variable lastKey = k. Then after ForEach use lastKey variable.
• "I did not dare to change the signature of OnNewHeader as I see there are multiple dependencies which might get broken" - no, only one. can do it later.

[AskAlexSharov]

Another way to print is subscribe to RPCDaemon as a client. For example:

wscat -c ws://localhost:8545
{"id": 1, "method": "eth_subscribe", "params": ["newHeads"]}

rpcdaemon must have --ws flag to enable websocker