Sharding support

.gitignore

@@ -6,3 +6,4 @@ aggregator.log
 /performance-test
 bft-config
 /data/
+.cache/

README.md

@@ -99,6 +99,14 @@ The service will start on `http://localhost:3000` by default.
 
 The service is configured via environment variables:
 
+### Chain Configuration
+
+| Variable        | Description     | Default   |
+|-----------------|-----------------|-----------|
+| `CHAIN_ID`      | Chain ID        | `unicity` |
+| `CHAIN_VERSION` | Chain version   | `1.0`     |
+| `CHAIN_FORK_ID` | Chain's Fork ID | `mainnet` |
+
 ### Server Configuration
 | Variable | Description | Default |
 |----------|-------------|---------|
@@ -220,6 +228,7 @@ Submit a state transition request to the aggregation layer with cryptographic va
 - `REQUEST_ID_MISMATCH` - RequestID doesn't match SHA256(publicKey || stateHash)
 - `INVALID_STATE_HASH_FORMAT` - StateHash not in proper DataHash imprint format
 - `INVALID_TRANSACTION_HASH_FORMAT` - TransactionHash not in proper DataHash imprint format
+- `INVALID_SHARD` - The commitment was sent to the wrong shard
 - `UNSUPPORTED_ALGORITHM` - Algorithm other than secp256k1
 
 #### `get_inclusion_proof`
@@ -551,6 +560,92 @@ The service implements a MongoDB-based leader election system:
 - **`follower`** - Processing API requests, monitoring for leadership
 - **`standalone`** - Single server mode (HA disabled)
 
+## Sharding
+
+To support horizontal scaling, the aggregators can be run in a sharded configuration consisting of one parent aggregator 
+and multiple child aggregators. In this mode, the global Sparse Merkle Tree (SMT) is split across the child nodes, and 
+agents must submit their commitments to the correct child node.
+
+For a more detailed technical explanation of the sharded SMT structure, please refer to the official specification:
+[https://github.com/unicitynetwork/specs/blob/main/smt.md](https://github.com/unicitynetwork/specs/blob/main/smt.md)
+
+### Commitment Routing
+
+The commitments are assigned to a shard based on the least significant bits of their commitment identifier. The number of 
+bits used to determine the shard is defined by the `SHARD_ID_LENGTH` configuration.
+
+For example `SHARD_ID_LENGTH: 1` means that the rightmost `1` bits of commitment identifier determines the correct 
+shard. In this case there would be 2 shards e.g. commitments ending with bit `0` would go to `shard-1`, and 
+commitments ending with bit `1` would go to the `shard-2`.
+
+In sharded setup only the parent aggregator talks to the BFT node.
+
+### Shard ID Encoding
+
+The `shardID` is a unique identifier for each shard that includes a `1` as its most significant bit (MSB). This prefix 
+bit ensures that the leading zeros are preserved for bit manipulations.
+
+Examples
+- For `SHARD_ID_LENGTH: 1` the valid `shardID`s are `0b10` (2) and `0b11` (3), for a total of two shards.
+- For `SHARD_ID_LENGTH: 2` the valid `shardID`s are `0b100` (4), `0b101` (5), `0b110` (6) and `0b111` (7), for a total of four shards.
+
+A child aggregator validates incoming commitments to ensure they belong to its shard. If a commitment is sent to the 
+wrong shard, the aggregator will reject it.
+
+### Example Sharded Setup
+
+The following diagram illustrates a sharded setup with one parent and two child aggregators for `SHARD_ID_LENGTH: 1`.
+
+```text
+          +--------------------+
+          | Parent Aggregator  |
+          | (2-leaf SMT)       |
+          +--------------------+
+           /                  \
+          /                    \
++----------------+     +----------------+
+| Child Agg. #1  |     | Child Agg. #2  |
+| ShardID = 0b10 |     | ShardID = 0b11 |
+| (handles *...0)|     | (handles *...1)|
++----------------+     +----------------+
+        ^                      ^
+        |                      |
++----------------+     +----------------+
+| Agent sends    |     | Agent sends    |
+| commitment     |     | commitment     |
+| ID = ...xxx0   |     | ID = ...xxx1   |
++----------------+     +----------------+
+```
+
+### Configuration
+
+The sharded setup is configured via environment variables, as seen in `sharding-compose.yml`.
+
+A **parent** aggregator is configured with:
+```yaml
+environment:
+  SHARDING_MODE: "parent"
+  SHARD_ID_LENGTH: 1
+```
+
+A **child** aggregator is configured with its unique `shardID` and the address of the parent, for example:
+
+Shard-1:
+```yaml
+environment:
+  SHARDING_MODE: "child"
+  SHARDING_CHILD_SHARD_ID: 2 # (binary 0b10)
+  SHARDING_CHILD_PARENT_RPC_ADDR: http://aggregator-root:3000
+```
+
+Shard-2:
+```yaml
+environment:
+  SHARDING_MODE: "child"
+  SHARDING_CHILD_SHARD_ID: 3 # (binary 0b11)
+  SHARDING_CHILD_PARENT_RPC_ADDR: http://aggregator-root:3000
+```
+
 ## Error Handling
 
 The service implements comprehensive JSON-RPC 2.0 error codes:

cmd/aggregator/main.go

@@ -102,41 +102,50 @@ func main() {
 	// Create the shared state tracker for block sync height
 	stateTracker := state.NewSyncStateTracker()
 
-	// Create the Round Manager
-	roundManager, err := round.NewRoundManager(ctx, cfg, log, commitmentQueue, storageInstance, stateTracker)
+	// Create round manager based on sharding mode
+	roundManager, err := round.NewManager(ctx, cfg, log, commitmentQueue, storageInstance, stateTracker)
 	if err != nil {
 		log.WithComponent("main").Error("Failed to create round manager", "error", err.Error())
 		gracefulExit(asyncLogger, 1)
 	}
 
-	// Perform initial SMT restoration. This is required in all modes.
+	// Initialize round manager (SMT restoration, etc.)
 	if err := roundManager.Start(ctx); err != nil {
 		log.WithComponent("main").Error("Failed to start round manager", "error", err.Error())
 		gracefulExit(asyncLogger, 1)
 	}
 
-	// Initialize HA Manager if enabled
+	// Initialize leader selector and HA Manager if enabled
+	var ls leaderSelector
 	var haManager *ha.HAManager
-	var leaderSelector *ha.LeaderElection
 	if cfg.HA.Enabled {
 		log.WithComponent("main").Info("High availability mode enabled")
-		leaderSelector = ha.NewLeaderElection(log, cfg.HA, storageInstance.LeadershipStorage())
-		leaderSelector.Start(ctx)
+		ls = ha.NewLeaderElection(log, cfg.HA, storageInstance.LeadershipStorage())
+		ls.Start(ctx)
+
+		// Disable block syncing for parent aggregator mode
+		// Parent mode uses state-based SMT (current shard roots) rather than history-based (commitment leaves)
+		disableBlockSync := cfg.Sharding.Mode == config.ShardingModeParent
+		if disableBlockSync {
+			log.WithComponent("main").Info("Block syncing disabled for parent aggregator mode - SMT will be reconstructed on leadership transition")
+		}
 
-		haManager = ha.NewHAManager(log, roundManager, leaderSelector, storageInstance, roundManager.GetSMT(), stateTracker, cfg.Processing.RoundDuration)
+		haManager = ha.NewHAManager(log, roundManager, ls, storageInstance, roundManager.GetSMT(), cfg.Sharding.Child.ShardID, stateTracker, cfg.Processing.RoundDuration, disableBlockSync)
 		haManager.Start(ctx)
-
 	} else {
 		log.WithComponent("main").Info("High availability mode is disabled, running as standalone leader")
-		// In non-HA mode, the node is always the leader, so activate it directly.
+		// In non-HA mode, activate the round manager directly
 		if err := roundManager.Activate(ctx); err != nil {
 			log.WithComponent("main").Error("Failed to activate round manager", "error", err.Error())
 			gracefulExit(asyncLogger, 1)
 		}
 	}
 
-	// Initialize service
-	aggregatorService := service.NewAggregatorService(cfg, log, roundManager, commitmentQueue, storageInstance, leaderSelector)
+	aggregatorService, err := service.NewService(ctx, cfg, log, roundManager, commitmentQueue, storageInstance, ls)
+	if err != nil {
+		log.WithComponent("main").Error("Failed to create service", "error", err.Error())
+		gracefulExit(asyncLogger, 1)
+	}
 
 	// Initialize gateway server
 	server := gateway.NewServer(cfg, log, aggregatorService)
@@ -166,17 +175,19 @@ func main() {
 		log.WithComponent("main").Error("Failed to stop server gracefully", "error", err.Error())
 	}
 
-	// Stop round manager
-	roundManager.Stop(shutdownCtx)
+	// Stop HA Manager if it was started
+	if haManager != nil {
+		haManager.Stop()
+	}
 
 	// Stop leader selector if it was started
-	if leaderSelector != nil {
-		leaderSelector.Stop(shutdownCtx)
+	if ls != nil {
+		ls.Stop(shutdownCtx)
 	}
 
-	// Stop HA Manager if it was started
-	if haManager != nil {
-		haManager.Stop()
+	// Stop round manager
+	if err := roundManager.Stop(shutdownCtx); err != nil {
+		log.WithComponent("main").Error("Failed to stop round manager gracefully", "error", err.Error())
 	}
 
 	// Close storage backends
@@ -194,3 +205,9 @@ func main() {
 		asyncLogger.Stop()
 	}
 }
+
+type leaderSelector interface {
+	IsLeader(ctx context.Context) (bool, error)
+	Start(ctx context.Context)
+	Stop(ctx context.Context)
+}