feat: WIP build PRT epoch commitment

Author: vfusco

internal/advancer/advancer.go

@@ -7,17 +7,15 @@ import (
 	"context"
 	"errors"
 	"fmt"
-	"net/http"
 	"os"
 	"path"
 	"strings"
 
-	"github.com/cartesi/rollups-node/internal/config"
-	"github.com/cartesi/rollups-node/internal/inspect"
 	"github.com/cartesi/rollups-node/internal/manager"
+	"github.com/cartesi/rollups-node/internal/merkle"
 	. "github.com/cartesi/rollups-node/internal/model"
 	"github.com/cartesi/rollups-node/internal/repository"
-	"github.com/cartesi/rollups-node/pkg/service"
+	pkgm "github.com/cartesi/rollups-node/pkg/machine"
 )
 
 var (
@@ -33,6 +31,7 @@ type AdvancerRepository interface {
 	ListInputs(ctx context.Context, nameOrAddress string, f repository.InputFilter, p repository.Pagination, descending bool) ([]*Input, uint64, error)
 	GetLastInput(ctx context.Context, appAddress string, epochIndex uint64) (*Input, error)
 	StoreAdvanceResult(ctx context.Context, appID int64, ar *AdvanceResult) error
+	UpdateEpochCommitment(ctx context.Context, appID int64, epochIndex uint64, commitment []byte) error
 	UpdateEpochsInputsProcessed(ctx context.Context, nameOrAddress string) (int64, error)
 	UpdateApplicationState(ctx context.Context, appID int64, state ApplicationState, reason *string) error
 	GetEpoch(ctx context.Context, nameOrAddress string, index uint64) (*Epoch, error)
@@ -41,90 +40,11 @@ type AdvancerRepository interface {
 	GetLastProcessedInput(ctx context.Context, appAddress string) (*Input, error)
 }
 
-// Service is the main advancer service that processes inputs through Cartesi machines
-type Service struct {
-	service.Service
-	snapshotsDir   string
-	repository     AdvancerRepository
-	machineManager manager.MachineProvider
-	inspector      *inspect.Inspector
-	HTTPServer     *http.Server
-	HTTPServerFunc func() error
-}
-
-// CreateInfo contains the configuration for creating an advancer service
-type CreateInfo struct {
-	service.CreateInfo
-	Config     config.AdvancerConfig
-	Repository repository.Repository
-}
-
-// Create initializes a new advancer service
-func Create(ctx context.Context, c *CreateInfo) (*Service, error) {
-	var err error
-	if err = ctx.Err(); err != nil {
-		return nil, err // This returns context.Canceled or context.DeadlineExceeded.
-	}
-
-	s := &Service{}
-	c.Impl = s
-
-	err = service.Create(ctx, &c.CreateInfo, &s.Service)
-	if err != nil {
-		return nil, err
-	}
-
-	s.repository = c.Repository
-	if s.repository == nil {
-		return nil, fmt.Errorf("repository on advancer service Create is nil")
-	}
-
-	// Create the machine manager
-	manager := manager.NewMachineManager(
-		ctx,
-		c.Repository,
-		s.Logger,
-		c.Config.FeatureMachineHashCheckEnabled,
-	)
-	s.machineManager = manager
-
-	// Initialize the inspect service if enabled
-	if c.Config.FeatureInspectEnabled {
-		s.inspector, s.HTTPServer, s.HTTPServerFunc = inspect.NewInspector(
-			c.Repository,
-			manager,
-			c.Config.InspectAddress,
-			c.LogLevel,
-			c.LogColor,
-		)
-	}
-
-	s.snapshotsDir = c.Config.SnapshotsDir
-
-	return s, nil
-}
-
-// Service interface implementation
-func (s *Service) Alive() bool     { return true }
-func (s *Service) Ready() bool     { return true }
-func (s *Service) Reload() []error { return nil }
-func (s *Service) Tick() []error {
-	if err := s.Step(s.Context); err != nil {
-		return []error{err}
-	}
-	return []error{}
-}
-func (s *Service) Stop(b bool) []error {
-	return nil
-}
-func (s *Service) Serve() error {
-	if s.inspector != nil && s.HTTPServerFunc != nil {
-		go s.HTTPServerFunc()
-	}
-	return s.Service.Serve()
-}
-func (s *Service) String() string {
-	return s.Name
+// temporary struct
+type EpochCommitment struct {
+	EpochIndex        uint64
+	EpochInputCount   uint64
+	CommitmentBuilder merkle.Builder
 }
 
 // getUnprocessedInputs retrieves inputs that haven't been processed yet
@@ -212,7 +132,7 @@ func (s *Service) processInputs(ctx context.Context, app *Application, inputs []
 			"index", input.Index)
 
 		// Advance the machine with this input
-		result, err := machine.Advance(ctx, input.RawData, input.Index)
+		result, err := machine.Advance(ctx, input.RawData, input.EpochIndex, input.Index, app.IsDaveConsensus())
 		if err != nil {
 			// If there's an error, mark the application as inoperable
 			s.Logger.Error("Error executing advance",
@@ -235,14 +155,16 @@ func (s *Service) processInputs(ctx context.Context, app *Application, inputs []
 
 			return err
 		}
-
+		// log advance result hashes
 		s.Logger.Info("Processing input finished",
 			"application", app.Name,
-			"epoch", input.EpochIndex,
-			"index", input.Index,
+			"epoch", result.EpochIndex,
+			"index", result.InputIndex,
 			"status", result.Status,
 			"outputs", len(result.Outputs),
 			"reports", len(result.Reports),
+			"hashes", len(result.Hashes),
+			"remaining_cycles", result.RemainingMetaCycles,
 		)
 
 		// Store the result in the database
@@ -266,18 +188,135 @@ func (s *Service) processInputs(ctx context.Context, app *Application, inputs []
 				// Continue processing even if snapshot creation fails
 			}
 		}
+
+		err = s.addCommimtmentLeafs(ctx, app, input, result)
+		if err != nil {
+			s.Logger.Error("Failed to add commitment leafs",
+				"application", app.Name,
+				"index", input.Index,
+				"error", err)
+		}
+
+		err = s.buildCommitment(ctx, app, input, machine)
+		if err != nil {
+			s.Logger.Error("Failed to build commitment",
+				"application", app.Name,
+				"index", input.Index,
+				"error", err)
+			return err
+		}
 	}
 
 	return nil
 }
 
-// handleEpochSnapshotAfterInputProcessed handles the snapshot creation after when an epoch is closed after an input was processed
-func (s *Service) handleEpochSnapshotAfterInputProcessed(ctx context.Context, app *Application) error {
-	// Check if the application has a epoch snapshot policy
-	if app.ExecutionParameters.SnapshotPolicy != SnapshotPolicy_EveryEpoch {
-		return nil
+func (s *Service) addCommimtmentLeafs(ctx context.Context, app *Application, input *Input, result *AdvanceResult) error {
+	if app.IsDaveConsensus() {
+		s.Logger.Debug("DaveConsensus: Adding input computation leafs",
+			"application", app.Name,
+			"epoch", input.EpochIndex,
+			"index", input.Index)
+		epochCommitment, exists := s.appCommitment[input.EpochApplicationID]
+		if !exists {
+			epochCommitment = &EpochCommitment{EpochIndex: input.EpochIndex}
+			s.appCommitment[input.EpochApplicationID] = epochCommitment
+		}
+		for i, hash := range result.Hashes {
+			s.Logger.Info("Advance result hash",
+				"application", app.Name,
+				"epoch", input.EpochIndex,
+				"index", input.Index,
+				"leaf_index", i,
+				"hash", fmt.Sprintf("%x", hash),
+				"repetitions", 1)
+			epochCommitment.CommitmentBuilder.Append(merkle.TreeLeaf(hash))
+		}
+		s.Logger.Info("Advance result hash",
+			"application", app.Name,
+			"epoch", input.EpochIndex,
+			"index", input.Index,
+			"leaf_count", len(result.Hashes),
+			"hash", result.MachineHash.String(),
+			"repetitions", result.RemainingMetaCycles)
+		epochCommitment.CommitmentBuilder.AppendRepeatedUint64(merkle.TreeLeaf(result.MachineHash), result.RemainingMetaCycles)
+		epochCommitment.EpochInputCount++
+	}
+	return nil
+}
+
+func (s *Service) buildCommitment(ctx context.Context, app *Application, input *Input, machine manager.MachineInstance) error {
+	if app.IsDaveConsensus() {
+		epochCommitment, exists := s.appCommitment[input.EpochApplicationID]
+		if !exists {
+			return nil
+		}
+		// If this is the last input and the epoch is closed, create a snapshot
+		isLastInput, err := s.isEpochLastInput(ctx, app, input)
+		if err != nil {
+			return err
+		}
+		if isLastInput {
+			s.Logger.Debug("DaveConsensus: Building commitment for last input of epoch",
+				"application", app.Name,
+				"epoch", input.EpochIndex,
+				"index", input.Index)
+			delete(s.appCommitment, input.EpochApplicationID)
+
+			remainingInputs := pkgm.InputsPerEpoch - epochCommitment.EpochInputCount
+			remainingStrides := remainingInputs << pkgm.Log2StridesPerInput
+			if remainingStrides > 0 {
+				machineHash, err := machine.Hash(ctx)
+				if err != nil {
+					return err
+				}
+
+				epochCommitment.CommitmentBuilder.AppendRepeatedUint64(merkle.TreeLeaf(machineHash), remainingStrides)
+			}
+
+			epochCommitmentTree := epochCommitment.CommitmentBuilder.Build()
+			commitment := epochCommitmentTree.GetRootHash()
+			s.Logger.Info("DaveConsensus: Epoch commitment built",
+				"application", app.Name,
+				"epoch", input.EpochIndex,
+				"commitment", fmt.Sprintf("%x", commitment))
+
+			err = s.repository.UpdateEpochCommitment(ctx, app.ID, input.EpochIndex, commitment.Bytes())
+			if err != nil {
+				return fmt.Errorf("failed to store epoch commitment: %w", err)
+			}
+		}
+	}
+	return nil
+}
+
+func (s *Service) isEpochLastInput(ctx context.Context, app *Application, input *Input) (bool, error) {
+	// Get the epoch for this input
+	epoch, err := s.repository.GetEpoch(ctx, app.IApplicationAddress.String(), input.EpochIndex)
+	if err != nil {
+		return false, fmt.Errorf("failed to get epoch: %w", err)
+	}
+
+	// Skip if the epoch is still open
+	if epoch.Status == EpochStatus_Open {
+		return false, nil
+	}
+
+	// Check if this is the last input of the epoch
+	lastInput, err := s.repository.GetLastInput(ctx, app.IApplicationAddress.String(), input.EpochIndex)
+	if err != nil {
+		return false, fmt.Errorf("failed to get last input: %w", err)
+	}
+
+	// If this is the last input and the epoch is closed, return true
+	if lastInput != nil && lastInput.Index == input.Index {
+		return true, nil
 	}
 
+	return false, nil
+}
+
+// handleEpochSnapshotAfterInputProcessed handles the snapshot creation after when an epoch is closed after an input was processed
+func (s *Service) handleEpochSnapshotAfterInputProcessed(ctx context.Context, app *Application) error {
 	// Get the machine instance for this application
 	machine, exists := s.machineManager.GetMachine(app.ID)
 	if !exists {
@@ -293,6 +332,19 @@ func (s *Service) handleEpochSnapshotAfterInputProcessed(ctx context.Context, ap
 	if lastProcessedInput == nil {
 		return nil
 	}
+	err = s.buildCommitment(ctx, app, lastProcessedInput, machine)
+	if err != nil {
+		s.Logger.Error("Failed to build commitment",
+			"application", app.Name,
+			"index", lastProcessedInput.Index,
+			"error", err)
+		return err
+	}
+
+	// Check if the application has a epoch snapshot policy
+	if app.ExecutionParameters.SnapshotPolicy != SnapshotPolicy_EveryEpoch {
+		return nil
+	}
 
 	// Handle the snapshot
 	return s.handleSnapshot(ctx, app, machine, lastProcessedInput)
@@ -314,25 +366,12 @@ func (s *Service) handleSnapshot(ctx context.Context, app *Application, machine
 
 	// For EVERY_EPOCH policy, check if this is the last input of the epoch
 	if policy == SnapshotPolicy_EveryEpoch {
-		// Get the epoch for this input
-		epoch, err := s.repository.GetEpoch(ctx, app.IApplicationAddress.String(), input.EpochIndex)
-		if err != nil {
-			return fmt.Errorf("failed to get epoch: %w", err)
-		}
-
-		// Skip if the epoch is still open
-		if epoch.Status == EpochStatus_Open {
-			return nil
-		}
-
-		// Check if this is the last input of the epoch
-		lastInput, err := s.repository.GetLastInput(ctx, app.IApplicationAddress.String(), input.EpochIndex)
+		// If this is the last input and the epoch is closed, create a snapshot
+		isLastInput, err := s.isEpochLastInput(ctx, app, input)
 		if err != nil {
-			return fmt.Errorf("failed to get last input: %w", err)
+			return err
 		}
-
-		// If this is the last input and the epoch is closed, create a snapshot
-		if lastInput != nil && lastInput.Index == input.Index {
+		if isLastInput {
 			return s.createSnapshot(ctx, app, machine, input)
 		}
 	}