retry on conflict 1

Author: miiu96

process/elasticproc/operations/serialize.go

@@ -30,7 +30,7 @@ func (op *operationsProcessor) prepareSerializedDataForAScResult(
 	index string,
 	selfShardID uint32,
 ) ([]byte, []byte, error) {
-	metaData := []byte(fmt.Sprintf(`{"update":{"_index":"%s","_id":"%s"}}%s`, index, converters.JsonEscape(scr.Hash), "\n"))
+	metaData := []byte(fmt.Sprintf(`{"update":{"_index":"%s","_id":"%s", "retry_on_conflict":3 }}%s`, index, converters.JsonEscape(scr.Hash), "\n"))
 	marshaledSCR, err := json.Marshal(scr)
 	if err != nil {
 		return nil, nil, err

process/elasticproc/operations/serialize_test.go

@@ -26,7 +26,7 @@ func TestOperationsProcessor_SerializeSCRS(t *testing.T) {
 	buffSlice := data.NewBufferSlice(data.DefaultMaxBulkSize)
 	err := op.SerializeSCRs(scrs, buffSlice, "operations", 0)
 	require.Nil(t, err)
-	require.Equal(t, `{"update":{"_index":"operations","_id":""}}
+	require.Equal(t, `{"update":{"_index":"operations","_id":"", "retry_on_conflict":3 }}
 {"script":{"source":"return"},"upsert":{"uuid":"","nonce":0,"gasLimit":0,"gasPrice":0,"value":"","valueNum":0,"sender":"","receiver":"","senderShard":0,"receiverShard":1,"prevTxHash":"","originalTxHash":"","callType":"","timestamp":0,"epoch":0}}
 { "index" : { "_index":"operations","_id" : "" } }
 {"uuid":"","nonce":0,"gasLimit":0,"gasPrice":0,"value":"","valueNum":0,"sender":"","receiver":"","senderShard":2,"receiverShard":0,"prevTxHash":"","originalTxHash":"","callType":"","timestamp":0,"epoch":0}

process/elasticproc/transactions/serialize.go

@@ -50,7 +50,7 @@ func (tdp *txsDatabaseProcessor) SerializeReceipts(receipts []*data.Receipt, buf
 // SerializeTransactionsFeeData will serialize transactions fee data
 func (tdp *txsDatabaseProcessor) SerializeTransactionsFeeData(txHashRefund map[string]*data.FeeData, buffSlice *data.BufferSlice, index string) error {
 	for txHash, feeData := range txHashRefund {
-		meta := []byte(fmt.Sprintf(`{"update":{ "_index":"%s","_id":"%s"}}%s`, index, converters.JsonEscape(txHash), "\n"))
+		meta := []byte(fmt.Sprintf(`{"update":{ "_index":"%s","_id":"%s", "retry_on_conflict":3}}%s`, index, converters.JsonEscape(txHash), "\n"))
 
 		var codeToExecute string
 		if feeData.GasRefunded != 0 {
@@ -142,7 +142,7 @@ func (tdp *txsDatabaseProcessor) SerializeTransactions(
 
 func serializeTxHashStatus(buffSlice *data.BufferSlice, txHashStatusInfo map[string]*outport.StatusInfo, index string) error {
 	for txHash, statusInfo := range txHashStatusInfo {
-		metaData := []byte(fmt.Sprintf(`{"update":{ "_index":"%s","_id":"%s"}}%s`, index, txHash, "\n"))
+		metaData := []byte(fmt.Sprintf(`{"update":{ "_index":"%s","_id":"%s", "retry_on_conflict":3}}%s`, index, txHash, "\n"))
 
 		newTx := &data.Transaction{
 			Status:         statusInfo.Status,
@@ -186,7 +186,7 @@ func prepareSerializedDataForATransaction(
 	selfShardID uint32,
 	index string,
 ) ([]byte, []byte, error) {
-	metaData := []byte(fmt.Sprintf(`{"update":{ "_index":"%s", "_id":"%s"}}%s`, index, converters.JsonEscape(tx.Hash), "\n"))
+	metaData := []byte(fmt.Sprintf(`{"update":{ "_index":"%s", "_id":"%s", "retry_on_conflict":3}}%s`, index, converters.JsonEscape(tx.Hash), "\n"))
 	marshaledTx, err := json.Marshal(tx)
 	if err != nil {
 		return nil, nil, err

process/elasticproc/transactions/serialize_test.go

@@ -104,7 +104,7 @@ func TestSerializeTransactionCrossShardTxSource(t *testing.T) {
 	}}, map[string]*outport.StatusInfo{}, 0, buffSlice, "transactions")
 	require.Nil(t, err)
 
-	expectedBuff := `{"update":{ "_index":"transactions", "_id":"txHash"}}
+	expectedBuff := `{"update":{ "_index":"transactions", "_id":"txHash", "retry_on_conflict":3}}
 {"script":{"source":"return"},"upsert":{"uuid":"","miniBlockHash":"","nonce":0,"round":0,"value":"","valueNum":0,"receiver":"","sender":"","receiverShard":1,"senderShard":0,"gasPrice":0,"gasLimit":0,"gasUsed":0,"fee":"","feeNum":0,"data":null,"signature":"","timestamp":0,"status":"","searchOrder":0,"version":1,"epoch":0}}
 `
 	require.Equal(t, expectedBuff, buffSlice.Buffers()[0].String())
@@ -145,7 +145,7 @@ func TestTxsDatabaseProcessor_SerializeTransactionWithRefund(t *testing.T) {
 	err := (&txsDatabaseProcessor{}).SerializeTransactionsFeeData(txHashRefund, buffSlice, "transactions")
 	require.Nil(t, err)
 
-	expectedBuff := `{"update":{ "_index":"transactions","_id":"txHash"}}
+	expectedBuff := `{"update":{ "_index":"transactions","_id":"txHash", "retry_on_conflict":3}}
 {"scripted_upsert": true, "script": {"source": "if ('create' == ctx.op) {ctx.op = 'noop'} else {ctx._source.fee = params.fee;ctx._source.feeNum = params.feeNum;ctx._source.gasUsed = params.gasUsed;}","lang": "painless","params": {"fee": "100000", "gasUsed": 5000, "feeNum": 5e-15, "gasRefunded": 0}},"upsert": {}}
 `
 	require.Equal(t, expectedBuff, buffSlice.Buffers()[0].String())