[subgraph] Fix Correctness of GDA Entities

packages/subgraph/config/hardhat.json

@@ -0,0 +1 @@
+{"network":"mainnet","testNetwork":"hardhat","hostStartBlock":0,"hostAddress":"0x4374EEcaAD0Dcaa149CfFc160d5a0552B1D092b0","cfaAddress":"0x44BF2a9217A2970A1bCC7529Bf1d40828C594320","idaAddress":"0x0826223156fF61F9abf620D2f58A06177DA664Cc","gdaAddress":"0x1Ad83f2789e013a41eA18819F0e24a4d41477b4c","superTokenFactoryAddress":"0x9b9389FacF9d345676b178EfEe328334cA711A38","resolverV1Address":"0x0e528E13E32cfB153818F1e10c7aff11ec5B46A3","nativeAssetSuperTokenAddress":"0x68bFD28e2AB62C450C563E1687f5f4F6b008149f","constantOutflowNFTAddress":"0x765FEFc8D7b2A6ec1D83626c50781e8cdd7e8284","constantInflowNFTAddress":"0x0983210c9f6f968ACC2010ac4dc56124dcDe2EC2"}

packages/subgraph/config/polygon-mainnet.json

@@ -0,0 +1 @@
+{"network":"matic","hostStartBlock":11650500,"hostAddress":"0x3E14dC1b13c488a8d5D310918780c983bD5982E7","cfaAddress":"0x6EeE6060f715257b970700bc2656De21dEdF074C","idaAddress":"0xB0aABBA4B2783A72C52956CDEF62d438ecA2d7a1","gdaAddress":"0x961dd5A052741B49B6CBf6759591f9D8576fCFb0","superTokenFactoryAddress":"0x2C90719f25B10Fc5646c82DA3240C76Fa5BcCF34","resolverV1Address":"0x8bDCb5613153f41B2856F71Bd7A7e0432F6dbe58","nativeAssetSuperTokenAddress":"0x3aD736904E9e65189c3000c7DD2c8AC8bB7cD4e3","constantOutflowNFTAddress":"0x554e2bbaCF43FD87417b7201A9F1649a3ED89d68","constantInflowNFTAddress":"0x55909bB8cd8276887Aae35118d60b19755201c68"}

packages/subgraph/schema.graphql

@@ -1815,6 +1815,11 @@ type Pool @entity {
     totalAmountInstantlyDistributedUntilUpdatedAt: BigInt!
     totalAmountFlowedDistributedUntilUpdatedAt: BigInt!
     totalAmountDistributedUntilUpdatedAt: BigInt!
+    totalFlowAdjustmentAmountDistributedUntilUpdatedAt: BigInt!
+
+    perUnitSettledValue: BigInt!
+    perUnitFlowRate: BigInt!
+
     """
     A member is any account which has more than 0 units in the pool.
     """
@@ -1870,6 +1875,9 @@ type PoolMember @entity {
     poolTotalAmountDistributedUntilUpdatedAt: BigInt!
     totalAmountReceivedUntilUpdatedAt: BigInt!
 
+    syncedPerUnitSettledValue: BigInt!
+    syncedPerUnitFlowRate: BigInt!
+
     account: Account!
     pool: Pool!
 

packages/subgraph/src/mappingHelpers.ts

@@ -36,7 +36,6 @@ import {
     getPoolDistributorID,
     getActiveStreamsDelta,
     getClosedStreamsDelta,
-    MAX_UINT256,
 } from "./utils";
 import { SuperToken as SuperTokenTemplate } from "../generated/templates";
 import { ISuperToken as SuperToken } from "../generated/templates/SuperToken/ISuperToken";
@@ -499,6 +498,11 @@ export function getOrInitPool(event: ethereum.Event, poolId: string): Pool {
         pool.totalAmountInstantlyDistributedUntilUpdatedAt = BIG_INT_ZERO;
         pool.totalAmountFlowedDistributedUntilUpdatedAt = BIG_INT_ZERO;
         pool.totalAmountDistributedUntilUpdatedAt = BIG_INT_ZERO;
+        pool.totalFlowAdjustmentAmountDistributedUntilUpdatedAt = BIG_INT_ZERO;
+
+        pool.perUnitSettledValue = BIG_INT_ZERO;
+        pool.perUnitFlowRate = BIG_INT_ZERO;
+
         pool.totalMembers = 0;
         pool.totalConnectedMembers = 0;
         pool.totalDisconnectedMembers = 0;
@@ -519,9 +523,6 @@ export function updatePoolTotalAmountFlowedAndDistributed(
     const timeDelta = event.block.timestamp.minus(pool.updatedAtTimestamp);
     const amountFlowedSinceLastUpdate = pool.flowRate.times(timeDelta);
 
-    pool.updatedAtBlockNumber = event.block.number;
-    pool.updatedAtTimestamp = event.block.timestamp;
-
     pool.totalAmountFlowedDistributedUntilUpdatedAt =
         pool.totalAmountFlowedDistributedUntilUpdatedAt.plus(
             amountFlowedSinceLastUpdate
@@ -554,13 +555,19 @@ export function getOrInitOrUpdatePoolMember(
         poolMember.totalAmountClaimed = BIG_INT_ZERO;
         poolMember.poolTotalAmountDistributedUntilUpdatedAt = BIG_INT_ZERO;
         poolMember.totalAmountReceivedUntilUpdatedAt = BIG_INT_ZERO;
-
+
+        poolMember.syncedPerUnitSettledValue = BIG_INT_ZERO;
+        poolMember.syncedPerUnitFlowRate = BIG_INT_ZERO;
+
         poolMember.account = poolMemberAddress.toHex();
         poolMember.pool = poolAddress.toHex();
     }
     poolMember.updatedAtTimestamp = event.block.timestamp;
     poolMember.updatedAtBlockNumber = event.block.number;
 
+    poolMember.updatedAtTimestamp = event.block.timestamp;
+    poolMember.updatedAtBlockNumber = event.block.number;
+
     return poolMember;
 }
 
@@ -1471,33 +1478,80 @@ export function updateAggregateEntitiesTransferData(
     tokenStatistic.save();
 }
 
+
+export function particleRTB(
+    perUnitSettledValue: BigInt,
+    perUnitFlowRate: BigInt,
+    currentTimestamp: BigInt,
+    lastUpdatedTimestamp: BigInt
+): BigInt {
+    const amountFlowedPerUnit = perUnitFlowRate.times(currentTimestamp.minus(lastUpdatedTimestamp));
+    return perUnitSettledValue.plus(amountFlowedPerUnit);
+}
+
+export function monetaryUnitPoolMemberRTB(pool: Pool, poolMember: PoolMember, currentTimestamp: BigInt): BigInt {
+    const poolPerUnitRTB = particleRTB(
+        pool.perUnitSettledValue,
+        pool.perUnitFlowRate,
+        currentTimestamp,
+        pool.updatedAtTimestamp
+    );
+    const poolMemberPerUnitRTB = particleRTB(
+        poolMember.syncedPerUnitSettledValue,
+        poolMember.syncedPerUnitFlowRate,
+        currentTimestamp,
+        poolMember.updatedAtTimestamp
+    );
+
+    return poolMember.totalAmountReceivedUntilUpdatedAt.plus(
+        poolPerUnitRTB.minus(poolMemberPerUnitRTB).times(poolMember.units)
+    );
+}
+
 /**
- * Updates `totalAmountReceivedUntilUpdatedAt` and `poolTotalAmountDistributedUntilUpdatedAt` fields
- * Requires an explicit save on the PoolMember entity.
- * Requires `pool.totalAmountDistributedUntilUpdatedAt` is updated *BEFORE* this function is called.
- * Requires that pool.totalUnits and poolMember.units are updated *AFTER* this function is called.
- * @param pool the pool entity
- * @param poolMember the pool member entity
- * @returns the updated pool member entity to be saved
+ * Updates the pool.perUnitSettledValue to the up to date value based on the current block,
+ * and updates the updatedAtTimestamp and updatedAtBlockNumber.
+ * @param pool pool entity
+ * @param block current block
+ * @returns updated pool entity
  */
-export function updatePoolMemberTotalAmountUntilUpdatedAtFields(pool: Pool, poolMember: PoolMember): PoolMember {
-    let amountReceivedDelta = BIG_INT_ZERO;
-    // if the pool member has any units, we calculate the delta
-    // otherwise the delta is going to be 0
-    if (!poolMember.units.equals(BIG_INT_ZERO)) {
-        const distributedAmountDelta = pool.totalAmountDistributedUntilUpdatedAt
-            .minus(poolMember.poolTotalAmountDistributedUntilUpdatedAt);
-
-        const isSafeToMultiplyWithoutOverflow = MAX_UINT256.div(poolMember.units).gt(distributedAmountDelta);
-        if (isSafeToMultiplyWithoutOverflow) {
-            amountReceivedDelta = distributedAmountDelta.times(poolMember.units).div(pool.totalUnits);
-        } else {
-            amountReceivedDelta = distributedAmountDelta.div(pool.totalUnits).times(poolMember.units);
-        }
-    }
-    poolMember.totalAmountReceivedUntilUpdatedAt =
-        poolMember.totalAmountReceivedUntilUpdatedAt.plus(amountReceivedDelta);
-    poolMember.poolTotalAmountDistributedUntilUpdatedAt = pool.totalAmountDistributedUntilUpdatedAt;
+export function settlePoolParticle(pool: Pool, block: ethereum.Block): Pool {
+    pool.perUnitSettledValue = particleRTB(
+        pool.perUnitSettledValue,
+        pool.perUnitFlowRate,
+        block.timestamp,
+        pool.updatedAtTimestamp
+    );
+    pool.updatedAtTimestamp = block.timestamp;
+    pool.updatedAtBlockNumber = block.number;
+
+    return pool;
+}
+
+export function settlePoolMemberParticle(poolMember: PoolMember, block: ethereum.Block): PoolMember {
+    poolMember.syncedPerUnitSettledValue = particleRTB(
+        poolMember.syncedPerUnitSettledValue,
+        poolMember.syncedPerUnitFlowRate,
+        block.timestamp,
+        poolMember.updatedAtTimestamp
+    );
+    poolMember.updatedAtTimestamp = block.timestamp;
+    poolMember.updatedAtBlockNumber = block.number;
+
+    return poolMember;
+}
+
+export function syncPoolMemberParticle(pool: Pool, poolMember: PoolMember): PoolMember {
+    poolMember.syncedPerUnitSettledValue = pool.perUnitSettledValue;
+    poolMember.syncedPerUnitFlowRate = pool.perUnitFlowRate;
+    poolMember.updatedAtTimestamp = pool.updatedAtTimestamp;
+    poolMember.updatedAtBlockNumber = pool.updatedAtBlockNumber;
 
     return poolMember;
 }
+
+export function settlePDPoolMemberMU(pool: Pool, poolMember: PoolMember, block: ethereum.Block): void {
+    pool = settlePoolParticle(pool, block);
+    poolMember.totalAmountReceivedUntilUpdatedAt = monetaryUnitPoolMemberRTB(pool, poolMember, block.timestamp);
+    poolMember = syncPoolMemberParticle(pool, poolMember);
+}

packages/subgraph/src/mappings/gdav1.ts

@@ -21,10 +21,11 @@ import {
     getOrInitPoolDistributor,
     getOrInitOrUpdatePoolMember,
     getOrInitTokenStatistic,
+    settlePDPoolMemberMU,
+    settlePoolParticle,
     updateATSStreamedAndBalanceUntilUpdatedAt,
     updateAggregateDistributionAgreementData,
     updatePoolDistributorTotalAmountFlowedAndDistributed,
-    updatePoolMemberTotalAmountUntilUpdatedAtFields,
     updatePoolTotalAmountFlowedAndDistributed,
     updateSenderATSStreamData,
     updateTokenStatisticStreamData,
@@ -33,6 +34,7 @@ import {
 import {
     BIG_INT_ZERO,
     createEventID,
+    divideOrZero,
     initializeEventEntity,
     membershipWithUnitsExists,
 } from "../utils";
@@ -98,7 +100,9 @@ export function handlePoolConnectionUpdated(
 
     // Update Pool Entity
     let pool = getOrInitPool(event, event.params.pool.toHex());
+    // @note we modify pool and poolMember here in memory, but do not save
     pool = updatePoolTotalAmountFlowedAndDistributed(event, pool);
+    settlePDPoolMemberMU(pool, poolMember, event.block);
     if (poolMember.units.gt(BIG_INT_ZERO)) {
         if (memberConnectedStatusUpdated) {
             // disconnected -> connected case
@@ -126,9 +130,6 @@ export function handlePoolConnectionUpdated(
             }
         }
     }
-
-    // Update totalAmountDistributedUntilUpdatedAt
-    poolMember = updatePoolMemberTotalAmountUntilUpdatedAtFields(pool, poolMember);
 
     pool.save();
     poolMember.save();
@@ -229,7 +230,12 @@ export function handleFlowDistributionUpdated(
 
     // Update Pool
     let pool = getOrInitPool(event, event.params.pool.toHex());
+
+    // @note that we are duplicating update of updatedAtTimestamp/BlockNumber here
+    // in the two functions
     pool = updatePoolTotalAmountFlowedAndDistributed(event, pool);
+    pool = settlePoolParticle(pool, event.block);
+    pool.perUnitFlowRate = divideOrZero(event.params.newDistributorToPoolFlowRate, pool.totalUnits);
     pool.flowRate = event.params.newTotalDistributionFlowRate;
     pool.adjustmentFlowRate = event.params.adjustmentFlowRate;
     pool.save();
@@ -312,7 +318,13 @@ export function handleInstantDistributionUpdated(
 
     // Update Pool
     let pool = getOrInitPool(event, event.params.pool.toHex());
+
+    // @note that we are duplicating update of updatedAtTimestamp/BlockNumber here
+    // in the two functions
     pool = updatePoolTotalAmountFlowedAndDistributed(event, pool);
+    pool = settlePoolParticle(pool, event.block);
+    // @note a speculations on what needs to be done
+    pool.perUnitSettledValue = pool.perUnitSettledValue.plus(divideOrZero(event.params.actualAmount, pool.totalUnits));
     const previousTotalAmountDistributed =
         pool.totalAmountDistributedUntilUpdatedAt;
     pool.totalAmountInstantlyDistributedUntilUpdatedAt =

packages/subgraph/src/mappings/superfluidPool.ts

@@ -9,9 +9,9 @@ import {
     _createTokenStatisticLogEntity,
     getOrInitPool,
     getOrInitOrUpdatePoolMember,
+    settlePDPoolMemberMU,
     updateATSStreamedAndBalanceUntilUpdatedAt,
     updateAggregateDistributionAgreementData,
-    updatePoolMemberTotalAmountUntilUpdatedAtFields,
     updatePoolTotalAmountFlowedAndDistributed,
     updateTokenStatsStreamedUntilUpdatedAt,
 } from "../mappingHelpers";
@@ -24,14 +24,17 @@ export function handleDistributionClaimed(event: DistributionClaimed): void {
 
     // Update Pool
     let pool = getOrInitPool(event, event.address.toHex());
-    pool = updatePoolTotalAmountFlowedAndDistributed(event, pool);
-    pool.save();
-
-    // Update PoolMember
     let poolMember = getOrInitOrUpdatePoolMember(event, event.address, event.params.member);
     poolMember.totalAmountClaimed = event.params.totalClaimed;
 
-    poolMember = updatePoolMemberTotalAmountUntilUpdatedAtFields(pool, poolMember);
+    // settle pool and pool member
+    pool = updatePoolTotalAmountFlowedAndDistributed(event, pool);
+    settlePDPoolMemberMU(pool, poolMember, event.block);
+
+    // Update PoolMember
+    poolMember.totalAmountClaimed = event.params.totalClaimed;
+
+    pool.save();
     poolMember.save();
 
     // Update Token Statistics
@@ -51,19 +54,37 @@ export function handleMemberUnitsUpdated(event: MemberUnitsUpdated): void {
     let pool = getOrInitPool(event, event.address.toHex());
     let poolMember = getOrInitOrUpdatePoolMember(event, event.address, event.params.member);
 
-    pool = updatePoolTotalAmountFlowedAndDistributed(event, pool);
-    poolMember = updatePoolMemberTotalAmountUntilUpdatedAtFields(pool, poolMember);
-
     const previousUnits = poolMember.units;
     const unitsDelta = event.params.newUnits.minus(previousUnits);
+    const newTotalUnits = pool.totalUnits.plus(unitsDelta);
+
+    pool = updatePoolTotalAmountFlowedAndDistributed(event, pool);
+    settlePDPoolMemberMU(pool, poolMember, event.block);
+
+    // @note TODO update the pool.perUnitFlowRate
+    // @note TODO update the poolMember.perUnitFlowRate
+    const existingPoolFlowRate = pool.perUnitFlowRate.times(pool.totalUnits);
+    let newPerUnitFlowRate: BigInt;
+    let remainderRate: BigInt;
+
+    if (!newTotalUnits.equals(BIG_INT_ZERO)) {
+        newPerUnitFlowRate = existingPoolFlowRate.div(newTotalUnits);
+        remainderRate = existingPoolFlowRate.minus(newPerUnitFlowRate.times(newTotalUnits));
+    } else {
+        remainderRate = existingPoolFlowRate;
+        newPerUnitFlowRate = BIG_INT_ZERO;
+    }
+    pool.perUnitFlowRate = newPerUnitFlowRate;
+    pool.totalUnits = newTotalUnits;
+
+    poolMember.syncedPerUnitFlowRate = poolMember.syncedPerUnitFlowRate.plus(remainderRate);
     poolMember.units = event.params.newUnits;
 
     if (poolMember.isConnected) {
         pool.totalConnectedUnits = pool.totalConnectedUnits.plus(unitsDelta);
     } else {
         pool.totalDisconnectedUnits = pool.totalDisconnectedUnits.plus(unitsDelta);
     }
-    pool.totalUnits = pool.totalUnits.plus(unitsDelta);
 
     // 0 units to > 0 units
     const didPoolMemberBecomeActive =  previousUnits.equals(BIG_INT_ZERO) && event.params.newUnits.gt(BIG_INT_ZERO)

packages/subgraph/src/utils.ts

@@ -387,3 +387,12 @@ export function createLogID(
         event.logIndex.toString()
     );
 }
+
+export function divideOrZero(
+    numerator: BigInt,
+    denominator: BigInt
+): BigInt {
+    return denominator.equals(BIG_INT_ZERO)
+        ? BIG_INT_ZERO
+        : numerator.div(denominator);
+}