Merge pull request #2434 from fluidity-money/develop

Prevent 0 rewards from taking place and new database changes

cmd/connector-common-winners-timescale/main.go

@@ -5,10 +5,17 @@
 package main
 
 import (
+	"github.com/fluidity-money/fluidity-app/lib/databases/postgres/solana"
 	database "github.com/fluidity-money/fluidity-app/lib/databases/timescale/winners"
 	queue "github.com/fluidity-money/fluidity-app/lib/queues/winners"
 )
 
 func main() {
-	queue.WinnersAll(database.InsertWinner)
+	go queue.WinnersEthereum(database.InsertWinner)
+
+	queue.WinnersSolana(func(winner queue.Winner) {
+		winningSignature := solana.GetIntermediateWinner(winner.TransactionHash)
+		winner.SendTransactionHash = winningSignature
+		database.InsertWinner(winner)
+	})
 }

cmd/connector-solana-amqp/main.go

@@ -142,7 +142,7 @@ func main() {
 			// choose the lowest block of any token
 			shouldBeLatestBlock := startingBlock == 0 || startingBlock > lastBlock
 
-			if shouldBeLatestBlock  {
+			if shouldBeLatestBlock {
 				startingBlock = lastBlock
 			}
 		}
@@ -188,7 +188,7 @@ func main() {
 
 // processSlot to write a block to the queue if it hasn't been already
 func processSlot(redisKey string, slot uint64) {
-	// if redisKey is set, caller is a websocket listener 
+	// if redisKey is set, caller is a websocket listener
 	if redisKey != "" {
 		redis.WriteLastBlock(redisKey, slot)
 	}

cmd/microservice-ethereum-worker-server/README.md

@@ -34,6 +34,16 @@ client.
 | `FLU_ETHEREUM_REDIS_APY_MOVING_AVERAGE_KEY` | Moving average key to use to track the APY with using Redis.                 |
 | `FLU_ETHEREUM_APPLICATION_CONTRACTS`        | List of comma-separated contract addresses to track as applications. (e.g. `0xae461ca67b15dc8dc81ce7615e0320da1a9ab8d5,0x3041cbd36888becc7bbcbc0045e3b1f144466f5f` ) |
 | `FLU_ETHEREUM_CHAINLINK_HTTP_URL`           | URL to use as the custom geth network for determining the price of ETH if the currently deployed network does not support it.                                        |
+| `FLU_ETHEREUM_GLOBAL_UTILITY_REWARDS`       | (<program name>,)+ is used to enable global utility rewards for each transfer. |
+
+## Notes
+
+### Global program application
+
+The global program (applications.ApplicationGlobalProgram) can be set
+to reward every FLUID transfer with an additional utility reward
+payout. FLU_ETHEREUM_GLOBAL_UTILITY_REWARDS is
+used for this.
 
 ## Building
 

cmd/microservice-ethereum-worker-server/main.go

@@ -19,6 +19,7 @@ import (
 	"github.com/fluidity-money/fluidity-app/common/ethereum/chainlink"
 	"github.com/fluidity-money/fluidity-app/common/ethereum/fluidity"
 
+	"github.com/fluidity-money/fluidity-app/lib/databases/postgres/failsafe"
 	worker_config "github.com/fluidity-money/fluidity-app/lib/databases/postgres/worker"
 	"github.com/fluidity-money/fluidity-app/lib/log"
 	"github.com/fluidity-money/fluidity-app/lib/queue"
@@ -78,6 +79,9 @@ const (
 
 	// EnvNetwork to differentiate between eth, arbitrum, etc
 	EnvNetwork = `FLU_ETHEREUM_NETWORK`
+
+	// EnvGlobalUtilityRewards to always pay out based on the internal config, just utility names.
+	EnvGlobalUtilityRewards = `FLU_ETHEREUM_GLOBAL_UTILITY_REWARDS`
 )
 
 type PayoutDetails struct {
@@ -101,6 +105,8 @@ func main() {
 		ammLpPoolAddr_ = mustEthereumAddressFromEnv(EnvAmmLpPoolAddress)
 
 		chainlinkEthPriceFeedUrl = os.Getenv(EnvChainlinkEthPriceNetworkUrl)
+
+		globalUtilityRewards_ = os.Getenv(EnvGlobalUtilityRewards)
 	)
 
 	ammLpPoolAddr := commonEth.ConvertGethAddress(ammLpPoolAddr_)
@@ -113,6 +119,17 @@ func main() {
 
 	tokenDetails := make(map[appTypes.UtilityName]token_details.TokenDetails)
 
+	// globalUtilityRewards addressed by the program name
+
+	globalUtilityRewards := make([]appTypes.UtilityName, 0)
+
+	for _, s := range strings.Split(globalUtilityRewards_, ",") {
+		if s == "" {
+			continue
+		}
+		globalUtilityRewards = append(globalUtilityRewards, appTypes.UtilityName(s))
+	}
+
 	for _, details := range strings.Split(tokenDetails_, ",") {
 		parts := strings.Split(details, ":")
 		if len(parts) != 3 {
@@ -188,6 +205,15 @@ func main() {
 		}
 	}
 
+	// these are the fluid clients that we build off later during our
+	// lookup, they need to include the base FLUID client as well as
+	// the clients for global utility rewards
+
+	baseFluidClients := append(
+		[]appTypes.UtilityName{appTypes.UtilityFluid},
+		globalUtilityRewards...,
+	)
+
 	queue.GetMessages(serverWorkAmqpTopic, func(message queue.Message) {
 		var hintedBlock worker.EthereumHintedBlock
 
@@ -449,8 +475,6 @@ func main() {
 				feePerTransfer.Quo(transactionFeeNormal, transferCountRat)
 			}
 
-			// for each fluid transfer, calculate the actual fee and chances
-
 			for _, transfer := range transaction.Transfers {
 				// if we have an application transfer, apply the fee
 
@@ -462,10 +486,33 @@ func main() {
 					logIndex          = transfer.LogIndex
 					appEmission       = transfer.AppEmissions
 
-					// the fluid token is always included
-					fluidClients = []appTypes.UtilityName{appTypes.UtilityFluid}
+					// the fluid token is always included (this implicitly copies the slice)
+					fluidClients = baseFluidClients
 				)
 
+				if logIndex == nil {
+					log.Fatal(func(k *log.Log) {
+						k.Format(
+							"Log index for transaction hash %v is nil!",
+							transactionHash,
+						)
+					})
+				}
+
+				// if the amount transferred was exactly 0, then we skip to the next transfer
+
+				if isDecoratedTransferZeroVolume(transfer) {
+					log.App(func(k *log.Log) {
+						k.Format(
+							"Skipped an empty amount transferred, hash %v, log index %v",
+							transfer.TransactionHash,
+							logIndex,
+						)
+					})
+
+					continue
+				}
+
 				senderAddress, senderAddressChanged := worker_config.LookupFeeSwitch(
 					senderAddress_,
 					dbNetwork,
@@ -476,6 +523,19 @@ func main() {
 					dbNetwork,
 				)
 
+				if senderAddress == recipientAddress {
+					log.App(func(k *log.Log) {
+						k.Format(
+							"Ignoring instance of sender and receiver being the same, skipping log index %v!",
+							logIndex,
+						)
+
+						k.Payload = senderAddress
+					})
+
+					continue
+				}
+
 				if senderAddressChanged {
 					emission.FeeSwitchSender.Reason = workerTypes.FeeSwitchReasonDatabase
 					emission.FeeSwitchSender.OriginalAddress = senderAddress_.String()
@@ -530,7 +590,7 @@ func main() {
 				pools, err := fluidity.GetUtilityVars(
 					gethClient,
 					registryAddress,
-					contractAddress,
+					contractAddress, // the token address
 					fluidClients,
 				)
 
@@ -544,7 +604,7 @@ func main() {
 				for _, pool := range pools {
 					// trigger
 					log.Debugf(
-						"Looking up the utility variables at registry %v, for the contract %v and the fluid clients %v, pool size native %v, token decimal scale %v, exchange rate %v, delta weight %v",
+						"Looked up the utility variables at registry %v, for the contract %v and the fluid clients %v, pool size native %v, token decimal scale %v, exchange rate %v, delta weight %v",
 						registryAddress,
 						contractAddress,
 						fluidClients,
@@ -556,7 +616,11 @@ func main() {
 				}
 
 				var (
-					normalPools  []workerTypes.UtilityVars
+					// normalPools are set during a normal transfer based on the TRF
+					normalPools []workerTypes.UtilityVars
+
+					// specialPools are set on a per-application basis based
+					// on the type of application used
 					specialPools []workerTypes.UtilityVars
 
 					// outputs from the trf
@@ -610,7 +674,8 @@ func main() {
 
 				payouts = append(payouts, normalPayout)
 
-				// special payouts!
+				// append special (utility client) payouts
+
 				for _, specialPool := range specialPools {
 					specialPayout := calculateSpecialPayoutDetails(
 						dbNetwork,
@@ -623,11 +688,17 @@ func main() {
 						secondsSinceLastEpoch,
 						emission,
 					)
+
 					payouts = append(payouts, specialPayout)
 				}
 
 				tokenDetails := fluidTokenDetails
 
+				// check if we've processed this before as a final failsafe before we submit
+				// the balls via a message and store an emission
+
+				failsafe.CommitTransactionHashIndex(transactionHash, *logIndex)
+
 				for _, payoutDetails := range payouts {
 
 					log.Debug(func(k *log.Log) {

cmd/microservice-ethereum-worker-server/special-payouts.go

@@ -17,7 +17,12 @@ import (
 	"github.com/fluidity-money/fluidity-app/lib/util"
 )
 
-func calculateSpecialPayoutDetails(dbNetwork network.BlockchainNetwork, pool workerTypes.UtilityVars, transferFeeNormal, currentAtx, payoutFreq *big.Rat, winningClasses, btx int, epochTime uint64, emission *worker.Emission) PayoutDetails {
+// calculateSpecialPayoutDetails by looking up the pool in the database
+// for overrides (and whether it's enabled), if it's not set then we
+// assume it's okay to continue, if the row exists, then we'll override
+// it. if it's set but it's disabled, then we return an empty struct.
+// logs the details via the emissions.
+func calculateSpecialPayoutDetails(dbNetwork network.BlockchainNetwork, pool workerTypes.UtilityVars, transferFeeNormal, currentAtx, payoutFreq *big.Rat, winningClasses, btx int, epochTime uint64, emission *worker.Emission) (specialPayout PayoutDetails) {
 	calculationType := pool.CalculationType
 
 	switch calculationType {
@@ -31,9 +36,12 @@ func calculateSpecialPayoutDetails(dbNetwork network.BlockchainNetwork, pool wor
 		)
 
 		// get overrides
-		details, found := workerDb.GetSpecialPoolOverrides(dbNetwork, pool.Name)
+		details, isEnabled, found := workerDb.GetSpecialPoolOverrides(dbNetwork, pool.Name)
+
+		switch {
+		case found && isEnabled:
+			// if the pool is enabled, then we need to handle the worker override behaviour
 
-		if found {
 			var (
 				winningClassesOverride = details.WinningClassesOverride
 				payoutFreqOverride     = details.PayoutFreqOverride
@@ -55,10 +63,30 @@ func calculateSpecialPayoutDetails(dbNetwork network.BlockchainNetwork, pool wor
 				pool.DeltaWeight.Set(deltaWeightOverride)
 				emission.SpecialPoolOptions.DeltaWeightOverride, _ = deltaWeightOverride.Float64()
 			}
+
+		case found:
+			// we found the pool, but it was disabled! so we're going to return an empty PayoutDetails!
+
+			log.App(func(k *log.Log) {
+				k.Format(
+					"Found pool %v, but was not enabled! Returning nothing on this utility client!",
+					pool.Name,
+				)
+			})
+
+			return
+
+		default:
+			log.App(func(k *log.Log) {
+				k.Format(
+					"Didn't find %v in the database, assuming it's enabled and that we don't want to override it",
+					pool.Name,
+				)
+			})
 		}
 
 		// call the trf normally now
-		specialPayout := calculatePayoutDetails(
+		specialPayout = calculatePayoutDetails(
 			workerTypes.TrfModeNoOptimisticSolution,
 			transferFeeNormal,
 			currentAtx,

cmd/microservice-ethereum-worker-server/util.go

@@ -19,12 +19,16 @@ import (
 	"github.com/fluidity-money/fluidity-app/lib/state"
 	"github.com/fluidity-money/fluidity-app/lib/types/ethereum"
 	"github.com/fluidity-money/fluidity-app/lib/types/misc"
+	typesWorker "github.com/fluidity-money/fluidity-app/lib/types/worker"
 	"github.com/fluidity-money/fluidity-app/lib/util"
 
 	ethCommon "github.com/ethereum/go-ethereum/common"
 	ethCrypto "github.com/ethereum/go-ethereum/crypto"
 )
 
+// ZeroInt that's used for empty comparisons, shouldn't mutate
+var ZeroInt = misc.BigIntFromInt64(0)
+
 func hexToPrivateKey(hex string) *ecdsa.PrivateKey {
 	privateKey, err := ethCrypto.HexToECDSA(hex)
 
@@ -187,3 +191,19 @@ func concatenatePastTransfers(blocks []uint64, transactionCounts []int) string {
 
 	return buf.String()
 }
+
+func isDecoratedTransferZeroVolume(transfer typesWorker.EthereumDecoratedTransfer) bool {
+	decorator := transfer.Decorator
+
+	if decorator == nil {
+		return true
+	}
+
+	volume := decorator.Volume
+
+	if volume == nil {
+		return true
+	}
+
+	return volume.Cmp(&ZeroInt.Int) == 0
+}