Fixes in syncer #52
Fixes in syncer #52
Conversation
|
|
||
| // Gets L1 (Arweave) interactions from the DB in batches | ||
| // Fills in last_sort_key for each interaction before emiting to the output channel | ||
| type ArweaveFetcher struct { |
There was a problem hiding this comment.
This name is kinda confusing to me - at first I thought that this code is responislbe for syncing/fetching data from the Arweave nodes.
For me the main purpose of this code is to fill the last sort keys for the already synced L1 transactions - so I would rename it to sth like ArweaveTxLastSortKeyFiller...
| err = self.db.Table(model.TableInteraction). | ||
| Where("block_height = ?", height). | ||
| Where("source=?", "arweave"). | ||
| Limit(self.Config.Forwarder.FetcherBatchSize). |
There was a problem hiding this comment.
Limit with Offset usually have some performance penalty...can't we simply use limit here and remember the last tx sortKey from the previous batch (and add sortKey > prevBatchLastSortKey condition) - which would allow to use the existing index on the sortKey column.
| Transaction(func(tx *gorm.DB) (err error) { | ||
| // Get a batch of L1 interactions | ||
| err = self.db.Table(model.TableInteraction). | ||
| Where("block_height = ?", height). |
There was a problem hiding this comment.
This again reminds me how I hate ORMs :-) can't we use raw sql queries here? :-)
| Limit(self.Config.Forwarder.FetcherBatchSize). | ||
| Offset(offset * self.Config.Forwarder.FetcherBatchSize). | ||
|
|
||
| // FIXME: -----------------> IS THIS ORDERING CORRECT? <----------------- |
| } | ||
|
|
||
| if state.FinishedBlockHeight < height { | ||
| state.FinishedBlockHeight = height |
There was a problem hiding this comment.
not sure I follow - how do we know at this point that we've processed ALL the L1 interactions for the given height and that FinishedBlockHeight can be safely set?
Shouldn't this be run (maybe it already is - but I don't see that in code) only for interactions from the last batch for the given height? which can be easilly verified (number of returned interactions < limit)
| }). | ||
| Error | ||
| if err != nil { | ||
| self.Log.WithError(err).Error("Failed to update stmonitorate after last block") |
There was a problem hiding this comment.
WTF stmonitorate is? :-) shouldn't it rather be 'FinishedBlockHeight' or sth similar
|
|
||
| // Update last sort key for each contract | ||
| for _, interaction := range interactions { | ||
| err = tx.Model(interaction). |
There was a problem hiding this comment.
is it possible to make a batch update here? instead of updating each row independently?
| // Omit those that are already in the lastSortKeys map | ||
| newContractIds := self.getNewContractIds(interactions, lastSortKeys) | ||
|
|
||
| // Get last sort key for each contract |
There was a problem hiding this comment.
those commens are sometimes kinda misleading :-) not for each, but for each NEW contract.
| return self | ||
| } | ||
|
|
||
| func (self *ArweaveFetcher) run() (err error) { |
There was a problem hiding this comment.
I would add some sanity check at the end of this function and verify that the newly set sortKey -> lastSortKey sequence is correct...at least for the first few weeks of running this in prod...
#40