[Raft] Add generic raftstore

pkg/aux_/store/raftstore/store.go

@@ -3,9 +3,12 @@ package raftstore
 import (
 	"context"
 
+	"github.com/interuss/dss/pkg/aux_/actions"
 	"github.com/interuss/dss/pkg/aux_/repos"
 	auxraftparams "github.com/interuss/dss/pkg/aux_/store/raftstore/params"
+	dsserr "github.com/interuss/dss/pkg/errors"
 	"github.com/interuss/dss/pkg/raftstore"
+	"github.com/interuss/dss/pkg/raftstore/consensus"
 	"github.com/interuss/stacktrace"
 	"go.uber.org/zap"
 )
@@ -18,5 +21,19 @@ func Init(ctx context.Context, logger *zap.Logger) (*raftstore.Store[repos.Repos
 	if err != nil {
 		return nil, stacktrace.Propagate(err, "failed to get aux raft parameters")
 	}
-	return raftstore.Init(ctx, logger.With(zap.String("service", "aux_")), params, func() repos.Repository { return &repo{} })
+	return raftstore.Init(ctx, logger.With(zap.String("service", "aux_")), params, &repo{}, actions.Registry)
+}
+
+func (r *repo) GetRepo() repos.Repository { return r }
+
+func (r *repo) GetSnapshot() ([]byte, error) {
+	return nil, stacktrace.NewErrorWithCode(dsserr.NotImplemented, "not implemented yet")
+}
+
+func (r *repo) RestoreFromSnapshot([]byte) error {
+	return stacktrace.NewErrorWithCode(dsserr.NotImplemented, "not implemented yet")
+}
+
+func (r *repo) Apply(_ context.Context, _ consensus.Proposal) (any, error) {
+	return nil, stacktrace.NewErrorWithCode(dsserr.NotImplemented, "not implemented yet")
 }

pkg/raftstore/consensus/consensus.go

@@ -110,7 +110,9 @@ func NewConsensus(ctx context.Context, logger *zap.Logger, connectParams params.
 	return consensus, nil
 }
 
+// TODO: ctx is currently ignored (see issue: https://github.com/interuss/dss/issues/1610)
 func (c *Consensus) Stop(ctx context.Context) {
+	// TODO: remove once (see issue: https://github.com/interuss/dss/issues/1610)
 	c.once.Do(func() {
 		c.logger.Info("stopping consensus")
 		close(c.stopC)
@@ -131,12 +133,8 @@ func (c *Consensus) Stop(ctx context.Context) {
 }
 
 // HandleClientRequest blocks until the proposal is committed and applied / dropped or until ctx is cancelled.
-func (c *Consensus) HandleClientRequest(ctx context.Context, requestType string, value any, readOnly bool) (any, error) {
-	proposal, err := c.newProposal(ctx, requestType, value, readOnly)
-	if err != nil {
-		return nil, stacktrace.Propagate(err, "failed to create proposal")
-	}
-
+func (c *Consensus) HandleClientRequest(ctx context.Context, requestType string, value []byte, readOnly bool) (any, error) {
+	proposal := c.newProposal(ctx, requestType, value, readOnly)
 	buf, err := json.Marshal(proposal)
 	if err != nil {
 		return nil, stacktrace.Propagate(err, "failed to marshal proposal")
@@ -223,6 +221,8 @@ func (c *Consensus) initTransport(ctx context.Context, nodeID uint64, clusterID
 // startRaftUpdatesConsumer starts a goroutine that processes the Ready channel of the Raft node and applies committed entries to the state machine
 func (c *Consensus) startRaftUpdatesConsumer(tickInterval time.Duration, snapshotInterval uint64) {
 	go func() {
+		// TODO: this shouldn't be triggered from inside the consensus instance, removing it will allow removing the once.
+		// (see issue: https://github.com/interuss/dss/issues/1610)
 		defer c.Stop(context.Background())
 
 		ticker := time.NewTicker(tickInterval)

pkg/raftstore/consensus/proposal.go

@@ -2,12 +2,11 @@ package consensus
 
 import (
 	"context"
-	"encoding/json"
 	"sync"
 	"time"
 
 	"github.com/google/uuid"
-	"github.com/interuss/stacktrace"
+	"github.com/interuss/dss/pkg/timestamp"
 )
 
 type EntryCommit struct {
@@ -30,21 +29,17 @@ type Proposal struct {
 	ReadOnly bool `json:"read_only"`
 }
 
-func (c *Consensus) newProposal(_ context.Context, requestType string, payload any, readOnly bool) (Proposal, error) {
-	// TODO - Fetch timestamp from context
-	value, err := json.Marshal(payload)
-	if err != nil {
-		return Proposal{}, stacktrace.Propagate(err, "failed to serialize proposal payload")
-	}
+func (c *Consensus) newProposal(ctx context.Context, requestType string, value []byte, readOnly bool) Proposal {
+	timestamp := timestamp.MustGetRequestTimestamp(ctx)
 
 	return Proposal{
 		ID:          uuid.NewString(),
 		NodeID:      c.nodeID,
-		Timestamp:   time.Now().UTC(),
+		Timestamp:   timestamp.UTC(),
 		RequestType: requestType,
 		Value:       value,
 		ReadOnly:    readOnly,
-	}, nil
+	}
 }
 
 type ProposalResult struct {

pkg/raftstore/store.go

@@ -3,45 +3,120 @@ package raftstore
 import (
 	"context"
 
+	"github.com/interuss/dss/pkg/logging"
 	"github.com/interuss/dss/pkg/raftstore/consensus"
 	raftparams "github.com/interuss/dss/pkg/raftstore/params"
 	"github.com/interuss/dss/pkg/store"
+	"github.com/interuss/dss/pkg/timestamp"
 	"github.com/interuss/stacktrace"
 	"go.uber.org/zap"
 )
 
+type RaftRepo[R any] interface {
+	GetRepo() R
+	// Apply is called on every committed entry. The proposal must be applied atomically.
+	// The any return mirrors store.OperationHandler.Execute: different requests yield different
+	// concrete result types. Callers recover the type via store.TransactWithResult.
+	Apply(ctx context.Context, proposal consensus.Proposal) (any, error)
+
+	// GetSnapshot returns a serialized view of current state, suitable
+	// for restoring via RestoreFromSnapshot.
+	GetSnapshot() ([]byte, error)
+
+	// RestoreFromSnapshot replaces all state with the snapshot in data.
+	// data is always the output of a prior GetSnapshot.
+	RestoreFromSnapshot(data []byte) error
+}
+
 type Store[R any] struct {
-	newRepo   func() R
-	consensus *consensus.Consensus
+	logger *zap.Logger
+
+	raftRepo RaftRepo[R]
+	cancel   context.CancelFunc
+	registry map[string]store.OperationHandler[R]
+
+	Consensus *consensus.Consensus
+
+	done chan struct{}
 }
 
-func Init[R any](ctx context.Context, logger *zap.Logger, params raftparams.ConnectParameters, newRepo func() R) (*Store[R], error) {
+func Init[R any](ctx context.Context, logger *zap.Logger, params raftparams.ConnectParameters, r RaftRepo[R], registry map[string]store.OperationHandler[R]) (*Store[R], error) {
+	ctx, cancel := context.WithCancel(ctx)
+
+	store := &Store[R]{
+		raftRepo: r,
+		logger:   logging.WithValuesFromContext(ctx, logger),
+		cancel:   cancel,
+		registry: registry,
+		done:     make(chan struct{}),
+	}
 	commitC := make(chan consensus.EntryCommit)
-	consensusInstance, err := consensus.NewConsensus(ctx, logger, params, func() ([]byte, error) { return nil, nil }, commitC)
+	go func() {
+		defer close(store.done)
+		store.processCommits(ctx, commitC)
+	}()
+
+	consensusInstance, err := consensus.NewConsensus(ctx, logger, params, r.GetSnapshot, commitC)
 	if err != nil {
 		return nil, stacktrace.Propagate(err, "failed to initialize consensus")
 	}
-	// TODO: start consumer goroutine reading from commitC
 
-	return &Store[R]{
-		newRepo:   newRepo,
-		consensus: consensusInstance,
-	}, nil
+	store.Consensus = consensusInstance
+
+	return store, nil
 }
 
 // Transact proposes the entry to Raft and blocks until it is committed and applied.
-func (s *Store[R]) Transact(_ context.Context, _ store.OperationRequest) (any, error) {
-	// TODO: implement
-	return nil, nil
+func (s *Store[R]) Transact(ctx context.Context, request store.OperationRequest) (any, error) {
+	handler, ok := s.registry[request.OperationID()]
+	if !ok {
+		return nil, stacktrace.NewError("no handler registered for operation %q", request.OperationID())
+	}
+	payload, err := handler.Encode(request)
+	if err != nil {
+		return nil, stacktrace.Propagate(err, "failed to encode op %q", request.OperationID())
+	}
+	return s.Consensus.HandleClientRequest(ctx, request.OperationID(), payload, handler.IsReadOnly)
 }
 
-// Interact returns a repository that can be used to query the store without proposing a Raft entry.
+// Interact returns the underlying Raft repo which, for every operation, will propose it to Raft and return the results.
 func (s *Store[R]) Interact(_ context.Context) (R, error) {
-	return s.newRepo(), nil
+	return s.raftRepo.GetRepo(), nil
 }
 
-// Close shuts down the consensus instance.
+// Close shuts down the consensus instance and processCommits loop.
+// TODO: pass a context to Stop then to consensus.Stop (see issue: https://github.com/interuss/dss/issues/1610).
 func (s *Store[R]) Close() error {
-	// TODO: implement
+	s.Consensus.Stop(context.Background())
+	s.cancel()
+	s.logger.Info("waiting for commit processing goroutine to exit")
+	<-s.done
 	return nil
 }
+
+// processCommits reads committed entries from the consensus layer and applies them via Apply.
+func (s *Store[R]) processCommits(ctx context.Context, commitCh <-chan consensus.EntryCommit) {
+	for {
+		select {
+		case <-ctx.Done():
+			s.logger.Info("stopping commit processing loop")
+			return
+		case commit, ok := <-commitCh:
+			if !ok {
+				s.logger.Info("commit channel closed, stopping commit processing loop")
+				return
+			}
+
+			if commit.SnapshotData != nil {
+				if err := s.raftRepo.RestoreFromSnapshot(commit.SnapshotData); err != nil {
+					s.logger.Fatal("failed to restore from snapshot", zap.Error(err))
+				}
+				continue
+			}
+
+			proposalCtx := timestamp.WithRequestTimestamp(ctx, commit.Prop.Timestamp)
+			result, err := s.raftRepo.Apply(proposalCtx, commit.Prop)
+			commit.Done <- consensus.ProposalResult{Result: result, Error: err}
+		}
+	}
+}

pkg/rid/store/raftstore/store.go

@@ -3,7 +3,10 @@ package raftstore
 import (
 	"context"
 
+	dsserr "github.com/interuss/dss/pkg/errors"
 	"github.com/interuss/dss/pkg/raftstore"
+	"github.com/interuss/dss/pkg/raftstore/consensus"
+	"github.com/interuss/dss/pkg/rid/actions"
 	"github.com/interuss/dss/pkg/rid/repos"
 	ridraftparams "github.com/interuss/dss/pkg/rid/store/raftstore/params"
 	"github.com/interuss/stacktrace"
@@ -18,5 +21,19 @@ func Init(ctx context.Context, logger *zap.Logger) (*raftstore.Store[repos.Repos
 	if err != nil {
 		return nil, stacktrace.Propagate(err, "failed to get rid raft parameters")
 	}
-	return raftstore.Init(ctx, logger.With(zap.String("service", "rid")), params, func() repos.Repository { return &repo{} })
+	return raftstore.Init(ctx, logger.With(zap.String("service", "rid")), params, &repo{}, actions.Registry)
+}
+
+func (r *repo) GetRepo() repos.Repository { return r }
+
+func (r *repo) GetSnapshot() ([]byte, error) {
+	return nil, stacktrace.NewErrorWithCode(dsserr.NotImplemented, "not implemented yet")
+}
+
+func (r *repo) RestoreFromSnapshot([]byte) error {
+	return stacktrace.NewErrorWithCode(dsserr.NotImplemented, "not implemented yet")
+}
+
+func (r *repo) Apply(_ context.Context, _ consensus.Proposal) (any, error) {
+	return nil, stacktrace.NewErrorWithCode(dsserr.NotImplemented, "not implemented yet")
 }

pkg/scd/store/raftstore/store.go

@@ -3,7 +3,10 @@ package raftstore
 import (
 	"context"
 
+	dsserr "github.com/interuss/dss/pkg/errors"
 	"github.com/interuss/dss/pkg/raftstore"
+	"github.com/interuss/dss/pkg/raftstore/consensus"
+	"github.com/interuss/dss/pkg/scd/actions"
 	"github.com/interuss/dss/pkg/scd/repos"
 	scdraftparams "github.com/interuss/dss/pkg/scd/store/raftstore/params"
 	"github.com/interuss/stacktrace"
@@ -18,5 +21,19 @@ func Init(ctx context.Context, logger *zap.Logger) (*raftstore.Store[repos.Repos
 	if err != nil {
 		return nil, stacktrace.Propagate(err, "failed to get scd raft parameters")
 	}
-	return raftstore.Init(ctx, logger.With(zap.String("service", "scd")), params, func() repos.Repository { return &repo{} })
+	return raftstore.Init(ctx, logger.With(zap.String("service", "scd")), params, &repo{}, actions.Registry)
+}
+
+func (r *repo) GetRepo() repos.Repository { return r }
+
+func (r *repo) GetSnapshot() ([]byte, error) {
+	return nil, stacktrace.NewErrorWithCode(dsserr.NotImplemented, "not implemented yet")
+}
+
+func (r *repo) RestoreFromSnapshot([]byte) error {
+	return stacktrace.NewErrorWithCode(dsserr.NotImplemented, "not implemented yet")
+}
+
+func (r *repo) Apply(_ context.Context, _ consensus.Proposal) (any, error) {
+	return nil, stacktrace.NewErrorWithCode(dsserr.NotImplemented, "not implemented yet")
 }

pkg/timestamp/timestamp.go

@@ -10,10 +10,10 @@ import (
 
 type timestampKey struct{}
 
-// RequestTimestampFromContext returns the request timestamp from the context, or an error if the value is not present or if it is zero.
+// requestTimestampFromContext returns the request timestamp from the context, or an error if the value is not present or if it is zero.
 // The timestamp is set by the Middleware when a query is received then (on the receiver side) by the Raftstore when the query is applied.
 // It is then used for deterministic execution of time-dependent queries.
-func RequestTimestampFromContext(ctx context.Context) (time.Time, error) {
+func requestTimestampFromContext(ctx context.Context) (time.Time, error) {
 	timestamp, ok := ctx.Value(timestampKey{}).(time.Time)
 	if !ok {
 		return time.Time{}, stacktrace.NewError("timestamp not found in context")
@@ -29,7 +29,7 @@ func RequestTimestampFromContext(ctx context.Context) (time.Time, error) {
 // MustGetRequestTimestamp returns the request timestamp from the context and panics if it is not
 // present or invalid, which is a programming error.
 func MustGetRequestTimestamp(ctx context.Context) time.Time {
-	timestamp, err := RequestTimestampFromContext(ctx)
+	timestamp, err := requestTimestampFromContext(ctx)
 	if err != nil {
 		panic(err)
 	}