fix: orhphaned jobs indexing

model/poaiEvents.go

@@ -14,7 +14,8 @@ type Allocation struct {
 	Id                 uint      `gorm:"primaryKey;autoIncrement" json:"id"`
 	AllocationCreation time.Time `json:"allocationCreation"`
 	BlockNumber        int64     `gorm:"type:bigint;not null" json:"blockNumber"`
-	TxHash             string    `gorm:"type:varchar(66);not null" json:"txHash"`
+	TxHash             string    `gorm:"type:varchar(66);not null;uniqueIndex:idx_allocation_tx_log,priority:1,where:log_index IS NOT NULL" json:"txHash"`
+	LogIndex           *uint     `gorm:"type:bigint;uniqueIndex:idx_allocation_tx_log,priority:2,where:log_index IS NOT NULL" json:"logIndex"`
 
 	JobId       string  `gorm:"type:text;not null" json:"jobId"`
 	JobName     string  `gorm:"type:text;default:null" json:"jobName"`

scripts/infuraFunctions.go

@@ -352,9 +352,11 @@ func decodeAllocLogs(vLog types.Log) (*model.Allocation, error) {
 		return nil, fmt.Errorf("jobId too large for uint64: %s", jobIDBig.String())
 	}
 	jobID = jobIDBig.Uint64()
+	logIndex := vLog.Index
 	result := model.Allocation{
 		CspAddress:  vLog.Address.String(),
 		TxHash:      vLog.TxHash.Hex(),
+		LogIndex:    &logIndex,
 		BlockNumber: int64(vLog.BlockNumber),
 
 		NodeAddress:     event.NodeAddress.String(),

service/statsService.go

@@ -23,6 +23,24 @@ import (
 
 const rpcRequestTimeout = 2 * time.Minute
 
+const (
+	orphanedAllocationCsp   = "0x0b81c24153bfbB2C98813da8ac0ef7e8b83ba389"
+	orphanedAllocationOwner = "0xE75981c3fb2734F263cEa7F81fF2Dd6586c1BF9A"
+)
+
+var orphanedAllocationBlocksByTx = map[string]int64{
+	"0x76c49d7f85eaabd39f421d892bd494e74786e476e348ced8e9b4e6564fbe18d1": 48325341,
+	"0x8de4ff5b4fd1c430038de25aa8526818c1890982654c53df62dbe930ab3f236a": 48368545,
+	"0x216d523066a93bfa697a1bb47a0f43ff82458b74b0d6c37a295c490323357703": 48411751,
+}
+
+var orphanedAllocationNodes = map[string]struct{}{
+	"0x3795d06dcd5cb35e25e669978e51c1c60c5105bc": {},
+	"0x3fa6b8254057670c8ac0673f3abffa277e07f88f": {},
+	"0xbca7a87c4730fb7e04e7858cd0152eea65b82492": {},
+	"0xca35c471dbb7296384ec448e8dc1757b0715200f": {},
+}
+
 func DailyGetStats() {
 	oldStats, err := storage.GetLatestStats()
 	if err != nil {
@@ -50,6 +68,9 @@ func DailyGetStats() {
 	}
 
 	from := oldStats.LastBlockNumber
+	if from > 0 {
+		from++
+	}
 	to, err := getChainLastBlockNumber()
 	if err != nil {
 		fmt.Println("error getting last block number: " + err.Error())
@@ -88,6 +109,11 @@ func DailyGetStats() {
 
 	for i, a := range allocEvents {
 		if owner, ok := nodeToOwner[a.NodeAddress]; ok {
+			if overrideOwner, overridden := orphanedAllocationOwnerOverride(a, owner); overridden {
+				owner = overrideOwner
+				fmt.Printf("Applied orphaned allocation owner override: tx=%s job=%s node=%s owner=%s\n",
+					a.TxHash, a.JobId, a.NodeAddress, owner)
+			}
 			allocEvents[i].UserAddress = owner
 		}
 	}
@@ -311,6 +337,21 @@ func DailyGetStats() {
 	manageEndingJobsAndSendEmails(allJobsDetails)
 }
 
+func orphanedAllocationOwnerOverride(allocation model.Allocation, resolvedOwner string) (string, bool) {
+	expectedBlock, ok := orphanedAllocationBlocksByTx[strings.ToLower(allocation.TxHash)]
+	if !ok || allocation.BlockNumber != expectedBlock {
+		return resolvedOwner, false
+	}
+	if !strings.EqualFold(allocation.CspAddress, orphanedAllocationCsp) {
+		return resolvedOwner, false
+	}
+	if _, ok := orphanedAllocationNodes[strings.ToLower(allocation.NodeAddress)]; !ok {
+		return resolvedOwner, false
+	}
+
+	return orphanedAllocationOwner, true
+}
+
 func getChainLastBlockNumber() (int64, error) {
 	client, err := ethclient.Dial(config.Config.Infura.ApiUrl + config.Config.Infura.Secret)
 	if err != nil {
@@ -526,9 +567,11 @@ func decodeAllocLogs(vLog types.Log) (*model.Allocation, error) {
 		return nil, fmt.Errorf("jobId too large for uint64: %s", jobIDBig.String())
 	}
 	jobID = jobIDBig.Uint64()
+	logIndex := vLog.Index
 	result := model.Allocation{
 		CspAddress:  vLog.Address.String(),
 		TxHash:      vLog.TxHash.Hex(),
+		LogIndex:    &logIndex,
 		BlockNumber: int64(vLog.BlockNumber),
 
 		NodeAddress: event.NodeAddress.String(),

service/statsService_test.go

@@ -2,9 +2,16 @@ package service
 
 import (
 	"fmt"
+	"math/big"
+	"strings"
 	"testing"
 
 	"github.com/NaeuralEdgeProtocol/ratio1-backend/config"
+	"github.com/NaeuralEdgeProtocol/ratio1-backend/model"
+	"github.com/NaeuralEdgeProtocol/ratio1-backend/ratio1abi"
+	"github.com/ethereum/go-ethereum/accounts/abi"
+	"github.com/ethereum/go-ethereum/common"
+	"github.com/ethereum/go-ethereum/core/types"
 	"github.com/stretchr/testify/require"
 )
 
@@ -19,3 +26,108 @@ func Test_GetDailyUsdcLocked(t *testing.T) {
 	require.Nil(t, err)
 	fmt.Println(value.String())
 }
+
+func TestOrphanedAllocationOwnerOverride(t *testing.T) {
+	matchingAllocation := model.Allocation{
+		BlockNumber: 48325341,
+		TxHash:      "0x76c49d7f85eaabd39f421d892bd494e74786e476e348ced8e9b4e6564fbe18d1",
+		NodeAddress: "0x3795d06dcd5cb35E25E669978e51c1C60c5105bC",
+		CspAddress:  "0x0b81c24153bFbB2C98813da8Ac0EF7e8b83Ba389",
+	}
+
+	tests := []struct {
+		name          string
+		allocation    model.Allocation
+		resolvedOwner string
+		expectedOwner string
+		overridden    bool
+	}{
+		{
+			name:          "overrides matching unresolved allocation",
+			allocation:    matchingAllocation,
+			resolvedOwner: "0x0000000000000000000000000000000000000000",
+			expectedOwner: orphanedAllocationOwner,
+			overridden:    true,
+		},
+		{
+			name:          "overrides matching allocation after node is relinked",
+			allocation:    matchingAllocation,
+			resolvedOwner: "0x1111111111111111111111111111111111111111",
+			expectedOwner: orphanedAllocationOwner,
+			overridden:    true,
+		},
+		{
+			name: "rejects wrong block",
+			allocation: func() model.Allocation {
+				allocation := matchingAllocation
+				allocation.BlockNumber++
+				return allocation
+			}(),
+			resolvedOwner: "0x0000000000000000000000000000000000000000",
+			expectedOwner: "0x0000000000000000000000000000000000000000",
+		},
+		{
+			name: "rejects unrelated node",
+			allocation: func() model.Allocation {
+				allocation := matchingAllocation
+				allocation.NodeAddress = "0x2222222222222222222222222222222222222222"
+				return allocation
+			}(),
+			resolvedOwner: "0x0000000000000000000000000000000000000000",
+			expectedOwner: "0x0000000000000000000000000000000000000000",
+		},
+	}
+
+	for _, test := range tests {
+		t.Run(test.name, func(t *testing.T) {
+			owner, overridden := orphanedAllocationOwnerOverride(test.allocation, test.resolvedOwner)
+			require.Equal(t, test.expectedOwner, owner)
+			require.Equal(t, test.overridden, overridden)
+		})
+	}
+}
+
+func TestOrphanedAllocationOwnerOverrideCoversAffectedEvents(t *testing.T) {
+	nodes := []string{
+		"0x3795d06dcd5cb35E25E669978e51c1C60c5105bC",
+		"0x3fA6b8254057670C8Ac0673F3ABFfa277E07f88f",
+		"0xBCA7A87C4730Fb7e04e7858CD0152eea65B82492",
+		"0xcA35C471Dbb7296384EC448e8Dc1757B0715200F",
+	}
+	for txHash, blockNumber := range orphanedAllocationBlocksByTx {
+		for _, nodeAddress := range nodes {
+			allocation := model.Allocation{
+				BlockNumber: blockNumber,
+				TxHash:      txHash,
+				NodeAddress: nodeAddress,
+				CspAddress:  orphanedAllocationCsp,
+			}
+			owner, overridden := orphanedAllocationOwnerOverride(allocation, common.Address{}.String())
+			require.True(t, overridden)
+			require.Equal(t, orphanedAllocationOwner, owner)
+		}
+	}
+}
+
+func TestDecodeAllocLogsPersistsLogIndex(t *testing.T) {
+	parsedABI, err := abi.JSON(strings.NewReader(ratio1abi.AllocationLogsAbi))
+	require.NoError(t, err)
+	event := parsedABI.Events["RewardsAllocatedV3"]
+	data, err := event.Inputs.NonIndexed().Pack(
+		common.HexToAddress("0x3795d06dcd5cb35E25E669978e51c1C60c5105bC"),
+		big.NewInt(318750),
+	)
+	require.NoError(t, err)
+
+	allocation, err := decodeAllocLogs(types.Log{
+		Address:     common.HexToAddress(orphanedAllocationCsp),
+		Topics:      []common.Hash{event.ID, common.BigToHash(big.NewInt(30))},
+		Data:        data,
+		TxHash:      common.HexToHash("0x76c49d7f85eaabd39f421d892bd494e74786e476e348ced8e9b4e6564fbe18d1"),
+		BlockNumber: 48325341,
+		Index:       563,
+	})
+	require.NoError(t, err)
+	require.NotNil(t, allocation.LogIndex)
+	require.Equal(t, uint(563), *allocation.LogIndex)
+}

storage/allocationStorer.go

@@ -5,6 +5,7 @@ import (
 
 	"github.com/NaeuralEdgeProtocol/ratio1-backend/model"
 	"gorm.io/gorm"
+	"gorm.io/gorm/clause"
 )
 
 func GetLatestAllocationBlock() (int64, error) {
@@ -31,15 +32,19 @@ func CreateAllocation(alloc *model.Allocation) error {
 		return err
 	}
 
-	txCreate := db.Create(&alloc)
+	txCreate := db.Clauses(clause.OnConflict{
+		Columns: []clause.Column{
+			{Name: "tx_hash"},
+			{Name: "log_index"},
+		},
+		TargetWhere: clause.Where{Exprs: []clause.Expression{
+			clause.Expr{SQL: "log_index IS NOT NULL"},
+		}},
+		DoNothing: true,
+	}).Create(&alloc)
 	if txCreate.Error != nil {
-		txCreate.Rollback()
 		return txCreate.Error
 	}
-	if txCreate.RowsAffected == 0 {
-		txCreate.Rollback()
-		return gorm.ErrRecordNotFound
-	}
 
 	return nil
 }

storage/allocationStorer_test.go

@@ -25,11 +25,11 @@ func TestGetAllocationsByJobIDsForJobDetails(t *testing.T) {
 	})
 
 	allocations := []model.Allocation{
-		allocationForJobDetailsTest(jobID1, "old job", model.JobType(1), "old project", 10, now.Add(-2*time.Hour)),
-		allocationForJobDetailsTest(jobID1, "latest job", model.JobType(2), "latest project", 11, now.Add(-time.Hour)),
-		allocationForJobDetailsTest(jobID1, "", model.JobType(3), "ignored project", 12, now),
-		allocationForJobDetailsTest(jobID2, "same block older", model.JobType(4), "same block old project", 20, now.Add(-30*time.Minute)),
-		allocationForJobDetailsTest(jobID2, "same block newer", model.JobType(5), "same block new project", 20, now.Add(-20*time.Minute)),
+		allocationForJobDetailsTest(jobID1, "old job", model.JobType(1), "old project", 10, 1, now.Add(-2*time.Hour)),
+		allocationForJobDetailsTest(jobID1, "latest job", model.JobType(2), "latest project", 11, 2, now.Add(-time.Hour)),
+		allocationForJobDetailsTest(jobID1, "", model.JobType(3), "ignored project", 12, 3, now),
+		allocationForJobDetailsTest(jobID2, "same block older", model.JobType(4), "same block old project", 20, 4, now.Add(-30*time.Minute)),
+		allocationForJobDetailsTest(jobID2, "same block newer", model.JobType(5), "same block new project", 20, 5, now.Add(-20*time.Minute)),
 	}
 	require.NoError(t, db.Create(&allocations).Error)
 
@@ -53,11 +53,12 @@ func TestGetAllocationsByJobIDsForJobDetailsEmptyInput(t *testing.T) {
 	require.Empty(t, result)
 }
 
-func allocationForJobDetailsTest(jobID, jobName string, jobType model.JobType, projectName string, blockNumber int64, allocationCreation time.Time) model.Allocation {
+func allocationForJobDetailsTest(jobID, jobName string, jobType model.JobType, projectName string, blockNumber int64, logIndex uint, allocationCreation time.Time) model.Allocation {
 	return model.Allocation{
 		AllocationCreation: allocationCreation,
 		BlockNumber:        blockNumber,
 		TxHash:             "0x0000000000000000000000000000000000000000000000000000000000000000",
+		LogIndex:           &logIndex,
 		JobId:              jobID,
 		JobName:            jobName,
 		JobType:            jobType,