diff --git a/pkg/aux_/store/raftstore/store.go b/pkg/aux_/store/raftstore/store.go index d09162d87..f74c1ccf0 100644 --- a/pkg/aux_/store/raftstore/store.go +++ b/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") } diff --git a/pkg/raftstore/consensus/consensus.go b/pkg/raftstore/consensus/consensus.go index d1c933370..a18427962 100644 --- a/pkg/raftstore/consensus/consensus.go +++ b/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) diff --git a/pkg/raftstore/consensus/proposal.go b/pkg/raftstore/consensus/proposal.go index 774992d65..006665001 100644 --- a/pkg/raftstore/consensus/proposal.go +++ b/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 { diff --git a/pkg/raftstore/store.go b/pkg/raftstore/store.go index 12bdb78bd..639d2e470 100644 --- a/pkg/raftstore/store.go +++ b/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} + } + } +} diff --git a/pkg/rid/store/raftstore/store.go b/pkg/rid/store/raftstore/store.go index 578aabeac..bc55a3dca 100644 --- a/pkg/rid/store/raftstore/store.go +++ b/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") } diff --git a/pkg/scd/store/raftstore/store.go b/pkg/scd/store/raftstore/store.go index a7eeb663a..799d6f1dd 100644 --- a/pkg/scd/store/raftstore/store.go +++ b/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") } diff --git a/pkg/timestamp/timestamp.go b/pkg/timestamp/timestamp.go index c88043589..28dbe9451 100644 --- a/pkg/timestamp/timestamp.go +++ b/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) }