feat(node): [NET-1455] Autostaker plugin

packages/node/bin/autostaker-manual-test.sh

@@ -0,0 +1,61 @@
+# Do not merge this manual test to main
+
+NODE_PRIVATE_KEY="1111111111111111111111111111111111111111111111111111111111111111"
+OWNER_PRIVATE_KEY="2222222222222222222222222222222222222222222222222222222222222222"
+SPONSORER_PRIVATE_KEY="3333333333333333333333333333333333333333333333333333333333333333"
+EARNINGS_PER_SECOND_1=1000
+EARNINGS_PER_SECOND_2=2000
+DELEGATED_AMOUNT=500000
+SPONSOR_AMOUNT=600000
+
+NODE_ADDRESS=$(ethereum-address $NODE_PRIVATE_KEY | jq -r '.address')
+OWNER_ADDRESS=$(ethereum-address $OWNER_PRIVATE_KEY | jq -r '.address')
+SPONSORER_ADDRESS=$(ethereum-address $SPONSORER_PRIVATE_KEY | jq -r '.address')
+
+cd ../../cli-tools
+
+echo 'Mint tokens'
+npx tsx bin/streamr.ts internal token-mint $NODE_ADDRESS 10000000 10000000 --env dev2
+npx tsx bin/streamr.ts internal token-mint $OWNER_ADDRESS 10000000 10000000 --env dev2
+
+echo 'Create operator'
+OPERATOR_CONTRACT_ADDRESS=$(npx tsx bin/streamr.ts internal operator-create -c 10 --node-addresses $NODE_ADDRESS --env dev2 --private-key $OWNER_PRIVATE_KEY | jq -r '.address') 
+npx tsx bin/streamr.ts internal operator-delegate $OPERATOR_CONTRACT_ADDRESS $DELEGATED_AMOUNT --env dev2 --private-key $OWNER_PRIVATE_KEY
+
+echo 'Create sponsorships'
+npx tsx bin/streamr.ts internal token-mint $SPONSORER_ADDRESS 10000000 10000000 --env dev2
+npx tsx bin/streamr.ts stream create /foo1 --env dev2 --private-key $SPONSORER_PRIVATE_KEY
+SPONSORSHIP_CONTRACT_ADDRESS_1=$(npx tsx bin/streamr.ts internal sponsorship-create /foo1 -e $EARNINGS_PER_SECOND_1 --env dev2 --private-key $SPONSORER_PRIVATE_KEY | jq -r '.address') 
+npx tsx bin/streamr.ts internal sponsorship-sponsor $SPONSORSHIP_CONTRACT_ADDRESS_1 $SPONSOR_AMOUNT --env dev2 --private-key $SPONSORER_PRIVATE_KEY
+npx tsx bin/streamr.ts stream create /foo2 --env dev2 --private-key $SPONSORER_PRIVATE_KEY
+SPONSORSHIP_CONTRACT_ADDRESS_2=$(npx tsx bin/streamr.ts internal sponsorship-create /foo2 -e $EARNINGS_PER_SECOND_2 --env dev2 --private-key $SPONSORER_PRIVATE_KEY | jq -r '.address') 
+npx tsx bin/streamr.ts internal sponsorship-sponsor $SPONSORSHIP_CONTRACT_ADDRESS_2 $SPONSOR_AMOUNT --env dev2 --private-key $SPONSORER_PRIVATE_KEY
+
+jq -n \
+    --arg nodePrivateKey "$NODE_PRIVATE_KEY" \
+    --arg operatorOwnerPrivateKey "$OWNER_PRIVATE_KEY" \
+    --arg operatorContractAddress "$OPERATOR_CONTRACT_ADDRESS" \
+    '{
+        "$schema": "https://schema.streamr.network/config-v3.schema.json",
+        client: {
+            auth: {
+                privateKey: $nodePrivateKey
+            },
+            environment: "dev2"
+        },
+        plugins: {
+            autostaker: {
+                operatorOwnerPrivateKey: $operatorOwnerPrivateKey,
+                operatorContractAddress: $operatorContractAddress
+            }
+        }
+    }' > ../node/configs/autostaker.json
+
+jq -n \
+    --arg operatorContract "$OPERATOR_CONTRACT_ADDRESS" \
+    --arg sponsorshipContract1 "$SPONSORSHIP_CONTRACT_ADDRESS_1" \
+    --arg sponsorshipContract2 "$SPONSORSHIP_CONTRACT_ADDRESS_2" \
+    '$ARGS.named'
+
+cd ../node
+npx tsx bin/streamr-node.ts configs/autostaker.json

packages/node/src/config/validateConfig.ts

@@ -9,6 +9,7 @@ export const validateConfig = (data: unknown, schema: Schema, contextName?: stri
     })
     addFormats(ajv)
     ajv.addFormat('ethereum-address', /^0x[a-zA-Z0-9]{40}$/)
+    ajv.addFormat('ethereum-private-key', /^(0x)?[a-zA-Z0-9]{64}$/)
     ajv.addSchema(DEFINITIONS_SCHEMA)
     if (!ajv.validate(schema, data)) {
         const prefix = (contextName !== undefined) ? (contextName + ': ') : ''

packages/node/src/pluginRegistry.ts

@@ -1,5 +1,6 @@
 import { Plugin } from './Plugin'
 import { StrictConfig } from './config/config'
+import { AutostakerPlugin } from './plugins/autostaker/AutostakerPlugin'
 import { ConsoleMetricsPlugin } from './plugins/consoleMetrics/ConsoleMetricsPlugin'
 import { HttpPlugin } from './plugins/http/HttpPlugin'
 import { InfoPlugin } from './plugins/info/InfoPlugin'
@@ -27,6 +28,8 @@ export const createPlugin = (name: string, brokerConfig: StrictConfig): Plugin<a
             return new SubscriberPlugin(name, brokerConfig)
         case 'info':
             return new InfoPlugin(name, brokerConfig)
+        case 'autostaker':
+            return new AutostakerPlugin(name, brokerConfig)
         default:
             throw new Error(`Unknown plugin: ${name}`)
     }

packages/node/src/plugins/autostaker/AutostakerPlugin.ts

@@ -0,0 +1,158 @@
+import { _operatorContractUtils, SignerWithProvider, StreamrClient } from '@streamr/sdk'
+import { collect, Logger, scheduleAtInterval, WeiAmount } from '@streamr/utils'
+import { Schema } from 'ajv'
+import { formatEther, Wallet } from 'ethers'
+import { Plugin } from '../../Plugin'
+import PLUGIN_CONFIG_SCHEMA from './config.schema.json'
+import { adjustStakes } from './payoutProportionalStrategy'
+import { Action, SponsorshipId, SponsorshipState } from './types'
+
+export interface AutostakerPluginConfig {
+    operatorContractAddress: string
+    // TODO is it possible implement this without exposing the private key here?
+    // e.g. by configuring so that operator nodes can stake behalf of the operator?
+    operatorOwnerPrivateKey: string
+    runIntervalInMs: number
+}
+
+const logger = new Logger(module)
+
+const getStakeOrUnstakeFunction = (action: Action): (
+    operatorOwnerWallet: SignerWithProvider,
+    operatorContractAddress: string,
+    sponsorshipContractAddress: string,
+    amount: WeiAmount
+) => Promise<void> => {
+    switch (action.type) {
+        case 'stake':
+            return _operatorContractUtils.stake
+        case 'unstake':
+            return _operatorContractUtils.unstake
+        default:
+            throw new Error('assertion failed')
+    }
+}
+
+export class AutostakerPlugin extends Plugin<AutostakerPluginConfig> {
+
+    private abortController: AbortController = new AbortController()
+
+    async start(streamrClient: StreamrClient): Promise<void> {
+        logger.info('Start autostaker plugin')
+        scheduleAtInterval(async () => {
+            try {
+                await this.runActions(streamrClient)
+            } catch (err) {
+                logger.warn('Error while running autostaker actions', { err })
+            }
+        }, this.pluginConfig.runIntervalInMs, false, this.abortController.signal)
+    }
+
+    private async runActions(streamrClient: StreamrClient): Promise<void> {
+        logger.info('Run autostaker actions')
+        const provider = (await streamrClient.getSigner()).provider
+        const operatorContract = _operatorContractUtils.getOperatorContract(this.pluginConfig.operatorContractAddress)
+            .connect(provider)
+        const stakedWei = await operatorContract.totalStakedIntoSponsorshipsWei()
+        const unstakedWei = (await operatorContract.valueWithoutEarnings()) - stakedWei
+        logger.info(`Balance: unstaked=${formatEther(unstakedWei)}, staked=${formatEther(stakedWei)}`)
+        const stakeableSponsorships = await this.getStakeableSponsorships(streamrClient)
+        logger.info(`Stakeable sponsorships: ${[...stakeableSponsorships.keys()].join(',')}`)
+        const stakes = await this.getStakes(streamrClient)
+        const stakeDescription = [...stakes.entries()].map(([sponsorshipId, amountWei]) => `${sponsorshipId}=${formatEther(amountWei)}`).join(', ')
+        logger.info(`Stakes before adjustments: ${stakeDescription}`)
+        const actions = adjustStakes({
+            operatorState: {
+                stakes,
+                unstakedWei
+            },
+            operatorConfig: {},  // TODO add maxSponsorshipCount
+            stakeableSponsorships,
+            environmentConfig: {
+                minimumStakeWei: 5000000000000000000000n  // TODO read from The Graph (network.minimumStakeWei)
+            }
+        })
+        const operatorOwnerWallet = new Wallet(this.pluginConfig.operatorOwnerPrivateKey, provider) as SignerWithProvider
+        for (const action of actions) {
+            logger.info(`Action: ${action.type} ${formatEther(action.amount)} ${action.sponsorshipId}`)
+            await getStakeOrUnstakeFunction(action)(operatorOwnerWallet,
+                this.pluginConfig.operatorContractAddress,
+                action.sponsorshipId,
+                action.amount
+            )
+        }
+    }
+
+    // eslint-disable-next-line class-methods-use-this
+    private async getStakeableSponsorships(streamrClient: StreamrClient): Promise<Map<SponsorshipId, SponsorshipState>> {
+        // TODO is there a better way to get the client? Maybe we should add StreamrClient#getTheGraphClient()
+        // TODO what are good where conditions for the sponsorships query so that we get all stakeable sponsorships
+        // but no non-stakables (e.g. expired)
+        // @ts-expect-error private
+        const queryResult = streamrClient.theGraphClient.queryEntities((lastId: string, pageSize: number) => {
+            return {
+                query: `
+                    {
+                        sponsorships(
+                            where:  {
+                                projectedInsolvency_gt: ${Math.floor(Date.now() / 1000)},
+                                spotAPY_gt: 0
+                                id_gt: "${lastId}"
+                            },
+                            first: ${pageSize}
+                        ) {
+                            id
+                            totalPayoutWeiPerSec
+                            totalStakedWei
+                        }
+                    }
+                `
+            }
+        })
+        const sponsorships: { id: SponsorshipId, totalPayoutWeiPerSec: bigint, totalStakedWei: bigint }[] = await collect(queryResult)
+        return new Map(sponsorships.map(
+            (sponsorship) => [sponsorship.id, {
+                totalPayoutWeiPerSec: BigInt(sponsorship.totalPayoutWeiPerSec),
+                totalStakedWei: BigInt(sponsorship.totalStakedWei)
+            }])
+        )
+    }
+
+    private async getStakes(streamrClient: StreamrClient): Promise<Map<SponsorshipId, bigint>> {
+        // TODO use StreamrClient#getTheGraphClient()
+        // @ts-expect-error private
+        const queryResult = streamrClient.theGraphClient.queryEntities((lastId: string, pageSize: number) => {
+            return {
+                query: `
+                    {
+                        stakes(
+                            where:  {
+                                operator: "${this.pluginConfig.operatorContractAddress.toLowerCase()}",
+                                id_gt: "${lastId}"
+                            },
+                            first: ${pageSize}
+                        ) {
+                            id
+                            sponsorship {
+                                id
+                            }
+                            amountWei
+                        }
+                    }
+                `
+            }
+        })
+        const stakes: { sponsorship: { id: SponsorshipId }, amountWei: bigint }[] = await collect(queryResult)
+        return new Map(stakes.map((stake) => [stake.sponsorship.id, BigInt(stake.amountWei) ]))
+    }
+
+    async stop(): Promise<void> {
+        logger.info('Stop autostaker plugin')
+        this.abortController.abort()
+    }
+
+    // eslint-disable-next-line class-methods-use-this
+    override getConfigSchema(): Schema {
+        return PLUGIN_CONFIG_SCHEMA
+    }
+}

packages/node/src/plugins/autostaker/config.schema.json

@@ -0,0 +1,29 @@
+{
+  "$id": "config.schema.json",
+  "$schema": "http://json-schema.org/draft-07/schema#",
+  "type": "object",
+  "description": "Autostaker plugin configuration",
+  "additionalProperties": false,
+  "required": [
+    "operatorContractAddress",
+    "operatorOwnerPrivateKey"
+  ],
+  "properties": {
+    "operatorContractAddress": {
+      "type": "string",
+      "description": "Operator contract Ethereum address",
+      "format": "ethereum-address"
+    },
+    "operatorOwnerPrivateKey": {
+      "type": "string",
+      "description": "Operator owner's private key",
+      "format": "ethereum-private-key"
+    },
+    "runIntervalInMs": {
+      "type": "integer",
+      "description": "The interval (in milliseconds) at which autostaking possibilities are analyzed and executed",
+      "minimum": 0,
+      "default": 30000
+    }
+  }
+}

packages/node/src/plugins/autostaker/payoutProportionalStrategy.ts

@@ -0,0 +1,113 @@
+import { sum } from '@streamr/utils'
+import partition from 'lodash/partition'
+import { Action, AdjustStakesFn } from './types'
+
+/**
+ * Allocate stake in proportion to the payout each sponsorship gives.
+ * Detailed allocation formula: each sponsorship should get the stake M + P / T,
+ *   where M is the environment-mandated minimum stake,
+ *         P is `totalPayoutWeiPerSec` of the considered sponsorship, and
+ *         T is sum of `totalPayoutWeiPerSec` of all sponsorships this operator stakes to.
+ * `totalPayoutWeiPerSec` is the total payout per second to all staked operators.
+ *
+ * In order to prevent that operators' actions cause other operators to change their plans,
+ *   this strategy only takes into account the payout, not what others have staked.
+ *
+ * In order to also prevent too much churn (joining and leaving),
+ *   this strategy will keep those sponsorships that they already have staked to, as long as they keep paying.
+ * Eventually sponsorships expire, and then the strategy will reallocate that stake elsewhere.
+ *
+ * This strategy tries to stake to as many sponsorships as it can afford.
+ * Since smaller amounts than the minimum stake can't be staked,
+ *   the tokens available to the operator limits the number of sponsorships they can stake to.
+ *
+ * Example:
+ * - there are sponsorships that pay out 2, 4, and 6 DATA/sec respectively.
+ * - minimum stake is 5000 DATA
+ * - operator has 11000 DATA (tokens in the contract plus currently staked tokens)
+ * 1. it only has enough for 2 sponsorships, so it chooses the ones that pay 4 and 6 DATA/sec
+ * 2. it allocates 5000 DATA to each of the 2 sponsorships (minimum stake)
+ * 3. it allocates the remaining 1000 DATA in proportion to the payout:
+ *   - 1000 * 4 / 10 = 400 DATA to the 4-paying sponsorship
+ *   - 1000 * 6 / 10 = 600 DATA to the 6-paying sponsorship
+ * - final target stakes are: 0, 5400, and 5600 DATA to the three sponsorships
+ * - the algorithm then outputs the stake and unstake actions to change the allocation to the target
+ **/
+export const adjustStakes: AdjustStakesFn = ({
+    operatorState,
+    operatorConfig,
+    stakeableSponsorships,
+    environmentConfig
+}): Action[] => {
+    const { unstakedWei, stakes } = operatorState
+    const { minimumStakeWei } = environmentConfig
+    const totalStakeableWei = sum([...stakes.values()]) + unstakedWei
+
+    if (Array.from(stakeableSponsorships.values()).find(({ totalPayoutWeiPerSec }) => totalPayoutWeiPerSec <= 0n)) {
+        throw new Error('payoutProportional: sponsorships must have positive totalPayoutWeiPerSec')
+    }
+
+    // find the number of sponsorships that we can afford to stake to
+    const targetSponsorshipCount = Math.min(
+        stakeableSponsorships.size,
+        operatorConfig.maxSponsorshipCount ?? Infinity,
+        Math.floor(Number(totalStakeableWei) / Number(minimumStakeWei)),
+    )
+
+    const sponsorshipList = Array.from(stakeableSponsorships.entries()).map(
+        ([id, { totalPayoutWeiPerSec }]) => ({ id, totalPayoutWeiPerSec })
+    )
+
+    // separate the stakeable sponsorships that we already have stakes in
+    const [
+        keptSponsorships,
+        potentialSponsorships,
+    ] = partition(sponsorshipList, ({ id }) => stakes.has(id))
+
+    // TODO: add secondary sorting of potentialSponsorships based on operator ID + sponsorship ID
+    // idea is: in case of a tie, operators should stake to different sponsorships
+
+    // pad the kept sponsorships to the target number, in the order of decreasing payout
+    potentialSponsorships.sort((a, b) => Number(b.totalPayoutWeiPerSec) - Number(a.totalPayoutWeiPerSec))
+    const selectedSponsorships = keptSponsorships
+        .concat(potentialSponsorships)
+        .slice(0, targetSponsorshipCount)
+
+    // calculate the target stakes for each sponsorship: minimum stake plus payout-proportional allocation
+    const minimumStakesWei = BigInt(selectedSponsorships.length) * minimumStakeWei
+    const payoutProportionalWei = totalStakeableWei - minimumStakesWei
+    const payoutSumWeiPerSec = sum(selectedSponsorships.map(({ totalPayoutWeiPerSec }) => totalPayoutWeiPerSec))
+
+    const targetStakes = new Map(payoutSumWeiPerSec > 0n ? selectedSponsorships.map(({ id, totalPayoutWeiPerSec }) =>
+        [id, minimumStakeWei + payoutProportionalWei * totalPayoutWeiPerSec / payoutSumWeiPerSec]) : [])
+
+    // calculate the stake differences for all sponsorships we have stakes in, or want to stake into
+    const sponsorshipIdList = Array.from(new Set([...stakes.keys(), ...targetStakes.keys()]))
+    const differencesWei = sponsorshipIdList.map((sponsorshipId) => ({
+        sponsorshipId,
+        differenceWei: (targetStakes.get(sponsorshipId) ?? 0n) - (stakes.get(sponsorshipId) ?? 0n)
+    })).filter(({ differenceWei: difference }) => difference !== 0n)
+
+    // TODO: filter out too small (TODO: decide what "too small" means) stakings and unstakings because those just waste gas
+
+    // sort the differences in ascending order (unstakings first, then stakings)
+    differencesWei.sort((a, b) => Number(a.differenceWei) - Number(b.differenceWei))
+
+    // force the net staking to equal unstakedWei (fixes e.g. rounding errors) by adjusting the largest staking (last in list)
+    const netStakingWei = sum(differencesWei.map(({ differenceWei: difference }) => difference))
+    if (netStakingWei !== unstakedWei && sponsorshipList.length > 0 && differencesWei.length > 0) {
+        const largestDifference = differencesWei.pop()!
+        largestDifference.differenceWei += unstakedWei - netStakingWei
+        // don't push back a zero difference
+        if (largestDifference.differenceWei !== 0n) {
+            differencesWei.push(largestDifference)
+        }
+    }
+
+    // convert differences to actions
+    return differencesWei.map(({ sponsorshipId, differenceWei }) => ({
+        type: differenceWei > 0n ? 'stake' : 'unstake',
+        sponsorshipId,
+        amount: differenceWei > 0n ? differenceWei : -differenceWei
+    }))
+}