Adopt snapshot streaming to handle snapshot fee balances

.github/workflows/pull_request.yml

@@ -17,7 +17,7 @@ jobs:
         with:
           path: tessellation
           repository: Constellation-Labs/tessellation
-          ref: v2.7.0
+          ref: v2.7.1
           token: ${{ secrets.GITHUB_TOKEN }}
 
       - name: Set up Java

.github/workflows/release.yml

@@ -21,7 +21,7 @@ jobs:
         with:
           path: tessellation
           repository: Constellation-Labs/tessellation
-          ref: v2.7.0
+          ref: v2.7.1
           token: ${{ secrets.GITHUB_TOKEN }}
 
       - name: Set up Java

project/Dependencies.scala

@@ -18,7 +18,7 @@ object Dependencies {
     val logstash = "7.2"
     val organizeImports = "0.6.0"
     val refined = "0.10.1"
-    val tessellation = "2.7.0"
+    val tessellation = "2.7.1"
     val weaver = "0.8.1"
   }
 

src/main/resources/application.conf

@@ -1,8 +1,9 @@
 snapshotStreaming {
   lastSnapshotPath = "lastSnapshot.json"
   lastIncrementalSnapshotPath = "lastIncrementalSnapshot.json"
-  environment = Testnet
-  collateral = 25000000000000
+  lastCurrencySnapshotsPath = "lastCurrencySnapshots"
+  environment = Dev
+  collateral = 0
   httpClient {
     timeout = "120s"
     idleTimeInPool = "30s"
@@ -31,8 +32,8 @@ snapshotStreaming {
     }
   }
   s3 {
-    bucketRegion = "us-west-1"
-    bucketName = "constellationlabs-testnet-snapshots"
-    bucketDir = "snapshot-streaming"
+    bucketRegion = ""
+    bucketName = ""
+    bucketDir = ""
   }
 }

src/main/scala/org/constellation/snapshotstreaming/Configuration.scala

@@ -33,6 +33,7 @@ class Configuration(sharedConfigReader: SharedConfigReader) {
 
   val lastFullSnapshotPath: Path = Path(config.getString("snapshotStreaming.lastSnapshotPath"))
   val lastIncrementalSnapshotPath: Path = Path(config.getString("snapshotStreaming.lastIncrementalSnapshotPath"))
+  val lastCurrencySnapshotsPath: Path = Path(config.getString("snapshotStreaming.lastCurrencySnapshotsPath"))
   val collateral: Amount = Amount(NonNegLong.unsafeFrom(config.getLong("snapshotStreaming.collateral")))
 
   val environment: AppEnvironment =

src/main/scala/org/constellation/snapshotstreaming/GlobalSnapshotContextService.scala

@@ -28,9 +28,6 @@ trait GlobalSnapshotContextService[F[_]] {
 
 object GlobalSnapshotContextService {
 
-  type CurrencySnapshotWithState =
-    Either[Signed[CurrencySnapshot], (Signed[CurrencyIncrementalSnapshot], CurrencySnapshotInfo)]
-
   def make[F[_]: Async: KryoSerializer: HasherSelector](
     globalSnapshotStateChannelEventsProcessor: GlobalSnapshotStateChannelEventsProcessor[F],
     globalSnapshotContextFns: GlobalSnapshotContextFunctions[F]
@@ -60,7 +57,7 @@ object GlobalSnapshotContextService {
                     }
                   })
                 }
-                .map(GlobalSnapshotWithState(artifact, newContext, _))
+                .map(GlobalSnapshotWithState(artifact, newContext, context.some, _))
             }
           }
 

src/main/scala/org/constellation/snapshotstreaming/SnapshotProcessor.scala

@@ -34,7 +34,11 @@ import fs2.Stream
 import org.constellation.snapshotstreaming.opensearch.OpensearchDAO
 import org.constellation.snapshotstreaming.opensearch.UpdateRequestBuilder
 import org.constellation.snapshotstreaming.s3.S3DAO
+import org.constellation.snapshotstreaming.storage.FileBasedLastFullSnapshotStorage
+import org.constellation.snapshotstreaming.storage.FileBasedLastIncrementalGlobalSnapshotStorage
+import org.constellation.snapshotstreaming.storage.LastCurrencySnapshotStorage
 import org.http4s.ember.client.EmberClientBuilder
+import org.tessellation.schema.GlobalSnapshot
 import org.tessellation.statechannel.StateChannelSnapshotBinary
 import org.typelevel.log4cats.slf4j.Slf4jLogger
 
@@ -71,11 +75,14 @@ object SnapshotProcessor {
           configuration.pullLimit.some,
           configuration.l0Peers.keys.some
         )
-      requestBuilder = UpdateRequestBuilder.make[F](configuration, txHasher: Hasher[F])
+      requestBuilder <- Resource.eval(UpdateRequestBuilder.make[F](configuration, txHasher: Hasher[F]))
       tesselationServices <- Resource.eval(
         TessellationServices.make[F](configuration)
       )
-      lastFullGlobalSnapshotStorage = FileBasedLastFullGlobalSnapshotStorage.make[F](configuration.lastFullSnapshotPath)
+      lastFullGlobalSnapshotStorage = FileBasedLastFullSnapshotStorage.make[F, GlobalSnapshot](
+        configuration.lastFullSnapshotPath
+      )
+      lastCurrencySnapshotStorage <- Resource.eval(LastCurrencySnapshotStorage.make[F](configuration.lastCurrencySnapshotsPath))
     } yield make(
       configuration,
       lastIncrementalGlobalSnapshotStorage,
@@ -84,26 +91,28 @@ object SnapshotProcessor {
       s3DAO,
       requestBuilder,
       tesselationServices,
-      lastFullGlobalSnapshotStorage
+      lastFullGlobalSnapshotStorage,
+      lastCurrencySnapshotStorage
     )
 
-  def make[F[_]: Async: KryoSerializer: JsonSerializer: HasherSelector](
+  def make[F[_]: Async: HasherSelector](
     configuration: Configuration,
     lastIncrementalGlobalSnapshotStorage: LastSnapshotStorage[F, GlobalIncrementalSnapshot, GlobalSnapshotInfo],
     l0Service: GlobalL0Service[F],
     opensearchDAO: OpensearchDAO[F],
     s3DAO: S3DAO[F],
     updateRequestBuilder: UpdateRequestBuilder[F],
     tessellationServices: TessellationServices[F],
-    lastFullGlobalSnapshotStorage: FileBasedLastFullGlobalSnapshotStorage[F]
+    lastFullGlobalSnapshotStorage: FileBasedLastFullSnapshotStorage[F, GlobalSnapshot],
+    lastCurrencySnapshotStorage: LastCurrencySnapshotStorage[F]
   ): SnapshotProcessor[F] = new SnapshotProcessor[F] {
     val logger = Slf4jLogger.getLogger[F]
 
     private def prepareAndExecuteBulkUpdate(
       globalSnapshotWithState: GlobalSnapshotWithState,
       hasher: Hasher[F]
     ): F[Unit] =
-      globalSnapshotWithState.pure[F].flatMap { case state @ GlobalSnapshotWithState(snapshot, _, _) =>
+      globalSnapshotWithState.pure[F].flatMap { case state @ GlobalSnapshotWithState(snapshot, _, _, _) =>
         Clock[F].realTime
           .map(d => new Date(d.toMillis))
           .flatMap(updateRequestBuilder.bulkUpdateRequests(state, _, hasher))
@@ -116,8 +125,19 @@ object SnapshotProcessor {
           .void
       }
 
+    private def setLastCurrencySnapshots(
+      snapshots: Map[Address, NonEmptyList[
+        Either[Hashed[
+          CurrencySnapshot
+        ], (Hashed[CurrencyIncrementalSnapshot], CurrencySnapshotInfo, Signed[StateChannelSnapshotBinary])]
+      ]],
+      hasher: Hasher[F]
+    ): F[Unit] = snapshots.toList.flatTraverse { case (addr, snapshots) =>
+      snapshots.traverse(s => lastCurrencySnapshotStorage.set(addr, s, hasher)).map(_.toList)
+    }.void
+
     private def process(globalSnapshotWithState: GlobalSnapshotWithState, hasher: Hasher[F]): F[Unit] =
-      globalSnapshotWithState.pure[F].flatMap { case state @ GlobalSnapshotWithState(snapshot, snapshotInfo, _) =>
+      globalSnapshotWithState.pure[F].flatMap { case state @ GlobalSnapshotWithState(snapshot, snapshotInfo, _, _) =>
         HasherSelector[F]
           .forOrdinal(snapshot.ordinal) { implicit hasher =>
             logger.info(
@@ -135,6 +155,7 @@ object SnapshotProcessor {
                 case Some(last) if Validator.isNextSnapshot(last, snapshot.signed.value) =>
                   s3DAO.uploadSnapshot(snapshot) >>
                     prepareAndExecuteBulkUpdate(state, hasher) >>
+                    setLastCurrencySnapshots(globalSnapshotWithState.currencySnapshots, hasher) >>
                     lastIncrementalGlobalSnapshotStorage.set(snapshot, snapshotInfo)
 
                 case Some(last) =>
@@ -151,6 +172,7 @@ object SnapshotProcessor {
                       case Some(last) if Validator.isNextSnapshot(last, snapshot.signed.value) =>
                         s3DAO.uploadSnapshot(snapshot) >>
                           prepareAndExecuteBulkUpdate(state, hasher) >>
+                          setLastCurrencySnapshots(globalSnapshotWithState.currencySnapshots, hasher) >>
                           lastIncrementalGlobalSnapshotStorage
                             .setInitial(snapshot, snapshotInfo)
                             .onError(e => logger.error(e)(s"Failure setting initial global snapshot!"))
@@ -210,7 +232,7 @@ object SnapshotProcessor {
                     .pullGlobalSnapshot(signedFullGlobalSnapshot.value.ordinal.next)
                     .map(
                       _.map(nextSnapshot =>
-                        GlobalSnapshotWithState(nextSnapshot, signedFullGlobalSnapshot.value.info, Map.empty)
+                        GlobalSnapshotWithState(nextSnapshot, signedFullGlobalSnapshot.value.info, None, Map.empty)
                       )
                     )
                     .map(_.toList)
@@ -223,13 +245,13 @@ object SnapshotProcessor {
           }
         }
         .evalTap { snapshots =>
-          snapshots.traverse { case GlobalSnapshotWithState(snapshot, _, _) =>
+          snapshots.traverse { case GlobalSnapshotWithState(snapshot, _, _, _) =>
             logger.info(s"Pulled following global snapshot: ${getSnapshotReference(snapshot).show}")
           }
         }
         .evalMap {
           _.tailRecM {
-            case (state @ GlobalSnapshotWithState(snapshot, _, _)) :: nextSnapshots
+            case (state @ GlobalSnapshotWithState(snapshot, _, _, _)) :: nextSnapshots
                 if configuration.terminalSnapshotOrdinal.forall(snapshot.ordinal <= _) =>
               val hasher = HasherSelector[F].getForOrdinal(snapshot.ordinal)
               process(state, hasher).as {
@@ -258,6 +280,7 @@ object SnapshotProcessor {
   case class GlobalSnapshotWithState(
     snapshot: Hashed[GlobalIncrementalSnapshot],
     snapshotInfo: GlobalSnapshotInfo,
+    prevSnapshotInfo: Option[GlobalSnapshotInfo],
     currencySnapshots: Map[Address, NonEmptyList[
       Either[Hashed[
         CurrencySnapshot