Feat/l1 msg queue #1055
Feat/l1 msg queue #1055
Conversation
|
so we are using approach 2? |
I think both are applicable here, because |
|
|
||
| for { | ||
| select { | ||
| case <-ms.ctx.Done(): |
There was a problem hiding this comment.
Why need to specifically handle <-ms.ctx.Done() and the subsequent select handle this again
| EndBlockHeader *types.Header | ||
| } | ||
|
|
||
| type MsgStorage struct { |
There was a problem hiding this comment.
add some metrics to MsgStorage eg:
- whether the task is running
- which head block is currently processed
| } | ||
|
|
||
| // PruneMessages deletes all messages that are older or equal to provided index | ||
| func (ms *MsgStorage) PruneMessages(lastIndex uint64) { |
There was a problem hiding this comment.
Who call this function? if no, maybe this it's called by itself periodically
rollup/l1/msg_storage.go
Outdated
| msgs: common.NewShrinkingMap[uint64, storedL1Message](1000), | ||
| reader: reader, | ||
| } | ||
| msgStorage.unsubscribeTracker = tracker.Subscribe(LatestChainHead, func(old, new []*types.Header) { |
There was a problem hiding this comment.
LatestChainHead should be configurable when we initialize the MsgStorage. This is essentially the time we wait to relay L1 messages from L1->L2.
rollup/l1/msg_storage.go
Outdated
| msgStorage.newChainNotifications = append(msgStorage.newChainNotifications, newChainNotification{old, new}) | ||
| }) | ||
|
|
||
| msgStorage.Start() |
There was a problem hiding this comment.
I think it's better to remove the start here and make it the responsibility of the user of MsgStorage to call Start. This way we can avoid asynchronicity in unit tests when testing MsgStorage
| var indexesToDelete []uint64 | ||
| for _, msg := range msgs { | ||
| contains := false | ||
| for _, header := range old { | ||
| if header.Hash() == msg.headerHash { | ||
| contains = true | ||
| break | ||
| } | ||
| } | ||
| if contains { | ||
| indexesToDelete = append(indexesToDelete, msg.l1msg.QueueIndex) | ||
| } | ||
| } | ||
| if len(indexesToDelete) > 0 { | ||
| ms.msgsMu.Lock() | ||
| for _, index := range indexesToDelete { | ||
| ms.msgs.Delete(index) | ||
| } | ||
| ms.msgsMu.Unlock() | ||
| } | ||
| } |
There was a problem hiding this comment.
How about storing an additional map L1 blockNum -> L1 message indexes. then we just need to iterate once and delete. this should be rare but it could be quite costly depending on how big msgs and old are
1. Purpose or design rationale of this PR
Remove existing
L1MessageQueuethat stores all msgs in db with new one thatL1Msgsfrom Finalized L1 blocks, but also load most recentL1Msgsand handle reorgs, crucial for reducing bridge latencyDesign explanation and task page
2. PR title
Your PR title must follow conventional commits (as we are doing squash merge for each PR), so it must start with one of the following types:
3. Deployment tag versioning
Has the version in
params/version.gobeen updated?4. Breaking change label
Does this PR have the
breaking-changelabel?