feat(ARCO-299): merge register and request transactions in blocktx #687
Conversation
|
| @@ -102,7 +102,7 @@ blocktx: | |||
| maxOpenConns: 80 # maximum open connections | |||
| sslMode: disable | |||
| recordRetentionDays: 28 # number of days for which data integrity is ensured | |||
| registerTxsInterval: 10s # time interval to read from the channel registered transactions | |||
| requestTxsInterval: 10s # time interval to register and request mined transactions | |||
There was a problem hiding this comment.
I think we should keep here the name registerTxsInterval and also topic RegisterTxTopic = "register-tx" and drop requestTxsInterval instead because a new tx is always registered. So that is kind a core functionality.
The functionality for requesting a tx because it wasn't mined for too long is rather an exception in case that e.g. blocktx was failing for some reason. So that is like an auxiliary functionality.
| @@ -681,92 +679,8 @@ func TestPostgresStore_UpsertBlockTransactions_CompetingBlocks(t *testing.T) { | |||
| require.NoError(t, err) | |||
|
|
|||
| // then | |||
| actual, err := sut.GetMinedTransactions(ctx, []*chainhash.Hash{txHash}) | |||
| actual, err := sut.UpsertAndGetMinedTransactions(ctx, [][]byte{txHash[:]}) | |||
There was a problem hiding this comment.
Does this really cover all of the test cases from TestPostgresStore_RegisterTransactions ?
Maybe we should keep and adapt TestPostgresStore_RegisterTransactions to TestPostgresStore_UpsertAndGetMinedTransactions
| func (p *Processor) StartProcessRequestTxs() { | ||
| p.waitGroup.Add(1) | ||
|
|
||
| txHashes := make([]*chainhash.Hash, 0, p.registerRequestTxsBatchSize) | ||
| txHashes := make([][]byte, 0, p.requestTxsBatchSize) |
| @@ -355,8 +305,8 @@ func (p *Processor) StartProcessRequestTxs() { | |||
| continue // retry, don't clear the txHashes slice | |||
| } | |||
|
|
|||
| txHashes = make([]*chainhash.Hash, 0, p.registerRequestTxsBatchSize) | |||
| ticker.Reset(p.registerRequestTxsInterval) | |||
| txHashes = make([][]byte, 0, p.requestTxsBatchSize) | |||
There was a problem hiding this comment.
| txHashes = make([][]byte, 0, p.requestTxsBatchSize) | |
| txHashes = txHashes [:0] |
| for _, minedTx := range minedTxs { | ||
| txBlock := &blocktx_api.TransactionBlock{ | ||
| TransactionHash: minedTx.TxHash, | ||
| BlockHash: minedTx.BlockHash, | ||
| BlockHeight: minedTx.BlockHeight, | ||
| MerklePath: minedTx.MerklePath, | ||
| } | ||
|
|
||
| err = p.mqClient.PublishMarshal(p.ctx, MinedTxsTopic, txBlock) |
There was a problem hiding this comment.
| err = p.mqClient.PublishMarshal(p.ctx, MinedTxsTopic, txBlock) | |
| publishErr = p.mqClient.PublishMarshal(p.ctx, MinedTxsTopic, txBlock) |
|
I have doubts about this solution. It might be beneficial to reduce one goroutine in In the current code, even if we fail to register a transaction and So, we don't need to add any new steps, fallbacks etc. to the current solution. Instead, we need to handle NATS connection errors. |
Description of Changes
Previously, there were separate goroutines for registering and requesting transactions in blocktx.
This caused an issue - when a transaction was not registered for some reason (e.g. nats queue temporarily down), we wouldn't get it back when requesting from metamorph, even though the transaction was mined and existing in blocktx database.
Now, with this PR, they are merged together, so that even if transaction would not be registered in the first call, it would be registered in any of the subsequent calls to
request-txtopic.Linked Issues / Tickets
Reference any related issues or tickets, e.g. "Closes #123".
Testing Procedure
Describe the tests you've added or any testing steps you've taken.
Checklist:
CHANGELOG.mdwith my changes