Implement pending join pool by TheMMaciek · Pull Request #1587 · Constellation-Labs/constellation

TheMMaciek · 2021-07-26T15:19:47Z

No description provided.

marcinwadon · 2021-07-27T10:05:11Z

schema/src/main/scala/org/constellation/schema/serialization/SchemaKryoRegistrar.scala

+        (NodeJoinsTheCluster.getClass, 209, CompatibleSerializer),
+        (NodeLeavesTheCluster.getClass, 210, CompatibleSerializer),
+        (NodeKickedOutByHealthcheck.getClass, 211, CompatibleSerializer),
+        (classOf[NextActiveNodes], 212, CompatibleSerializer),


why not classOf[] as it was used before?

Can you elaborate?

I mean, you used Foo.getClass instead of classOf[Foo] as it was used almost everywhere else. that's why I'm asking ;)

Ok you mean e.g. the line 121. That's how we got the class for the case objects. classOf[] doesn't work for case objects AFAIR

yup. Ok, didn't know

marcinwadon · 2021-07-27T10:06:08Z

schema/src/main/scala/org/constellation/schema/snapshot/Snapshot.scala

+  lastSnapshot: String,
+  checkpointBlocks: Seq[String],
+  publicReputation: SortedMap[Id, Double],
+  nextActiveNodes: NextActiveNodes,


what if we will make a rollback and there are no such nodes in the cluster?

Rollback case may not be handled yet. Essentially during rollback the NextActiveNodes should be ignored and we should start from the initial nodes - currently based on the whitelisting file. I mentioned it on the daily sometime ago.

I guess it will be then NextActiveNodes(Set.empty, Set.empty) but @TheMMaciek please confirm

During rollback these will be set to some nodes, but we need to ignore them during rollback and use the initialFullNodes -> currently these are first 3 nodes in the whitelisting file. The initial nodes logic can be changed of course, especially if we will remove whitelisting.

marcinwadon · 2021-07-27T10:08:07Z

src/main/scala/org/constellation/consensus/ConsensusManager.scala

  keyPair: KeyPair,
  checkpointsQueueInstance: DataResolverCheckpointsEnqueue[F]
-) {
+)(implicit F: Concurrent[F], CS: ContextShift[F], T: Timer[F]) {


you need F only explicitly

I guess we wanted to have it uniform, so I moved all of them to this notation. I can move only what I need.

marcinwadon · 2021-07-27T10:14:32Z

src/main/scala/org/constellation/domain/cluster/ClusterStorageAlgebra.scala

+  def getActivePeersIds(withSelfId: Boolean = false): F[Set[Id]]
+  def getActivePeers: F[Map[Id, PeerData]]
+  def getActiveLightPeersIds(withSelfId: Boolean = false): F[Set[Id]]
+  def getActiveLightPeers(): F[Map[Id, PeerData]]


getActiveLightPeers instead of getActiveLightPeers() to be consistent with the other methods. if it makes an action, then (). if not and it is only a get, no ()

That's exactly what happens, getActiveLightPeers() performs an action.

You can look it up, but essentially what it does is it takes the list of lightPeersIds, and fetches their PeerData from the list of all peers. There is some more logic happening for this case and I'm not sure it should stay their inside a storage, but that's how I rebased it initially to test if the implementation still works. The thing is this function is needed in different places around the code base, it was previously in Cluster class, now it's in storage and it kind of clashes with the idea that storage should only store data so possibly I will need to improve it.

but all in all, it's get active light peers. doesn't matter how it works under the hood. if it takes it from the database, API or Mars ;)

Yeah kind of makes sense ;). I remember we discussed that only for a simple getter it should be parameterless (just getting the value) but here much more happens actually. Though it's true that behind the interface it's not distinguishable. That said I can change it to parameterless and we can assume that every get function with no parameters should be in parameterless notation, or we could drop that convention and use empty parameter list simply everywhere.

marcinwadon · 2021-07-27T10:15:33Z

src/main/scala/org/constellation/domain/cluster/ClusterStorageAlgebra.scala

+  def getActiveFullPeersIds(withSelfId: Boolean = false): F[Set[Id]]
+  def getActiveFullPeers(): F[Map[Id, PeerData]]
+  def setActiveFullPeers(peers: Set[Id]): F[Unit]
+  def isAnActivePeer: F[Boolean]


There is node and cluster. Node storage is about self-knowledge. Cluster storage about the other peers.
Here, isAnActivePeer takes no parameter so I think it should belong to node storage, not cluster storage.

Sure, it still can be moved to where it fits more.

marcinwadon · 2021-07-27T10:18:40Z

src/main/scala/org/constellation/domain/redownload/RedownloadService.scala

+  // TODO: notify new light nodes to join the pool also and update your light nodes list
+  // TODO: we need a consensus over new light and full nodes between current active full nodes
+  private def notifyNewActivePeersAndLeaveThePool(): F[Unit] =
+    for {
+      _ <- logger.debug(s"Notifying new active peers to join the active pool!")


Why this is a part of redownload service? It should be separated

I guess it can be moved. Redownload flow seemed like the best place to put it initially.

imo it should be in a separate service. and the same for storage related to joining pool

marcinwadon · 2021-07-27T10:19:40Z

src/main/scala/org/constellation/infrastructure/cluster/ClusterStorageInterpreter.scala


  private val peers: Ref[F, Map[Id, PeerData]] = Ref.unsafe(Map.empty[Id, PeerData])

+  // TODO: move initialization of the refs below to somewhere outside - Cluster class startup?


It's in startup flow currently. It was initialized during Cluster class startup till refs where moved to storage classes and because clusterStorage and nodeStorage are only aware of themselves the initial state (like initial full nodes set based on whitelisting file) can't be set during storage initial creation. I did put keyPair or nodeId in some storages but I'm not convinced it should be like that, so I still plan to move these outside of storage and handle logic that needs these parameters differently.

marcinwadon suggested changes Jul 27, 2021

View reviewed changes

Implement joining pool

68b663d

TheMMaciek force-pushed the implement-pending-join-pool-REBASED branch from 40f825b to 68b663d Compare August 11, 2021 13:54


		private val peers: Ref[F, Map[Id, PeerData]] = Ref.unsafe(Map.empty[Id, PeerData])

		// TODO: move initialization of the refs below to somewhere outside - Cluster class startup?

Conversation

TheMMaciek commented Jul 26, 2021

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

TheMMaciek Jul 27, 2021 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

TheMMaciek Jul 27, 2021 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

3 participants

TheMMaciek Jul 27, 2021 •

edited

Loading

TheMMaciek Jul 27, 2021 •

edited

Loading