diff --git a/cmds/core-service/main.go b/cmds/core-service/main.go index 33e9e5922..19701412e 100644 --- a/cmds/core-service/main.go +++ b/cmds/core-service/main.go @@ -24,6 +24,7 @@ import ( auxs "github.com/interuss/dss/pkg/aux_/store" "github.com/interuss/dss/pkg/build" dsserr "github.com/interuss/dss/pkg/errors" + requestlocality "github.com/interuss/dss/pkg/locality" "github.com/interuss/dss/pkg/logging" "github.com/interuss/dss/pkg/rid/application" rid_v1 "github.com/interuss/dss/pkg/rid/server/v1" @@ -100,7 +101,7 @@ func createAuxServer(ctx context.Context, locality string, publicEndpoint string return nil, stacktrace.NewError("Public endpoint not set") } - auxStore, err := auxs.Init(ctx, logger, true) + auxStore, err := auxs.Init(ctx, logger, true, locality) if err != nil { return nil, err } @@ -121,7 +122,7 @@ func createAuxServer(ctx context.Context, locality string, publicEndpoint string func createRIDServers(ctx context.Context, locality string, logger *zap.Logger) (*rid_v1.Server, *rid_v2.Server, error) { - ridStore, err := rids.Init(ctx, logger, true) + ridStore, err := rids.Init(ctx, logger, true, locality) if err != nil { return nil, nil, err } @@ -151,9 +152,9 @@ func createRIDServers(ctx context.Context, locality string, logger *zap.Logger) }, nil } -func createSCDServer(ctx context.Context, logger *zap.Logger) (*scd.Server, error) { +func createSCDServer(ctx context.Context, logger *zap.Logger, locality string) (*scd.Server, error) { - scdStore, err := scds.Init(ctx, logger, true) + scdStore, err := scds.Init(ctx, logger, true, locality) if err != nil { return nil, err } @@ -350,7 +351,7 @@ func RunHTTPServer(ctx context.Context, ctxCanceler func(), address, locality st // Initialize strategic conflict detection if *enableSCD { - scdV1Server, err = createSCDServer(ctx, logger) + scdV1Server, err = createSCDServer(ctx, logger, locality) if err != nil { return stacktrace.Propagate(err, "Failed to create strategic conflict detection server") } @@ -365,6 +366,7 @@ func RunHTTPServer(ctx context.Context, ctxCanceler func(), address, locality st handler = http.TimeoutHandler(handler, *timeout, "request timeout") handler = logging.HTTPMiddleware(logger, *dumpRequests, handler) handler = timestamp.RequestTimestampMiddleware(handler) + handler = requestlocality.LocalityMiddleware(locality)(handler) if *enableMetrics || *enableTracing { // We use the default settings; the APIRouter handler will override the span value accordingly, as it has more information. diff --git a/cmds/db-manager/cleanup/evict.go b/cmds/db-manager/cleanup/evict.go index 6cdf70d62..c03c0b7a9 100644 --- a/cmds/db-manager/cleanup/evict.go +++ b/cmds/db-manager/cleanup/evict.go @@ -55,12 +55,12 @@ func evict(cmd *cobra.Command, _ []string) error { logger := logging.WithValuesFromContext(ctx, logging.Logger) - scdStore, err := scds.Init(ctx, logger, false) + scdStore, err := scds.Init(ctx, logger, false, *locality) if err != nil { return err } - ridStore, err := rids.Init(ctx, logger, false) + ridStore, err := rids.Init(ctx, logger, false, *locality) if err != nil { return err } diff --git a/pkg/aux_/store/raftstore/store.go b/pkg/aux_/store/raftstore/store.go index 3c1fd3c33..8f8a4270f 100644 --- a/pkg/aux_/store/raftstore/store.go +++ b/pkg/aux_/store/raftstore/store.go @@ -28,7 +28,7 @@ type repo struct { memRepo repos.Repository } -func Init(ctx context.Context, logger *zap.Logger) (*raftstore.Store[repos.Repository], error) { +func Init(ctx context.Context, logger *zap.Logger, locality string) (*raftstore.Store[repos.Repository], error) { params, err := auxraftparams.GetConnectParameters() if err != nil { return nil, stacktrace.Propagate(err, "failed to get aux raft parameters") @@ -40,7 +40,7 @@ func Init(ctx context.Context, logger *zap.Logger) (*raftstore.Store[repos.Repos } r := &repo{memStore: memStore, memRepo: memStore.GetRepo()} - store, err := raftstore.Init(ctx, logger.With(zap.String("service", "aux_")), params, r, nil) + store, err := raftstore.Init(ctx, logger.With(zap.String("service", "aux_")), locality, params, r, nil) if err != nil { return nil, stacktrace.Propagate(err, "failed to initialize aux raftstore") } diff --git a/pkg/aux_/store/store.go b/pkg/aux_/store/store.go index 2532842a7..93148ff80 100644 --- a/pkg/aux_/store/store.go +++ b/pkg/aux_/store/store.go @@ -19,12 +19,12 @@ import ( type Store = dssstore.Store[repos.Repository] // Init selects and initializes the aux store backend. -func Init(ctx context.Context, logger *zap.Logger, withCheckCron bool) (Store, error) { +func Init(ctx context.Context, logger *zap.Logger, withCheckCron bool, locality string) (Store, error) { switch storeType := params.GetStoreParameters().StoreType; storeType { case params.SQLStoreType: return auxsqlstore.Init(ctx, logger, withCheckCron) case params.RaftStoreType: - return auxraftstore.Init(ctx, logger) + return auxraftstore.Init(ctx, logger, locality) case params.MemStoreType: return auxmemstore.Init(ctx, logger) default: diff --git a/pkg/locality/locality.go b/pkg/locality/locality.go new file mode 100644 index 000000000..8ea51faa8 --- /dev/null +++ b/pkg/locality/locality.go @@ -0,0 +1,36 @@ +package locality + +import ( + "context" + "net/http" + + "github.com/interuss/stacktrace" +) + +type localityKey struct{} + +// MustGetRequestLocality returns the request locality from the context and panics if it is not +// present, which is a programming error. +func MustGetRequestLocality(ctx context.Context) string { + locality, ok := ctx.Value(localityKey{}).(string) + if !ok { + panic(stacktrace.NewError("request locality not present in context")) + } + + return locality +} + +// WithRequestLocality returns a new context with the given locality. +func WithRequestLocality(ctx context.Context, locality string) context.Context { + return context.WithValue(ctx, localityKey{}, locality) +} + +// LocalityMiddleware is an HTTP middleware that stamps each incoming request with this +// DSS instance's locality so that locality-dependent operations execute deterministically across nodes. +func LocalityMiddleware(locality string) func(http.Handler) http.Handler { + return func(next http.Handler) http.Handler { + return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + next.ServeHTTP(w, r.WithContext(WithRequestLocality(r.Context(), locality))) + }) + } +} diff --git a/pkg/raftstore/consensus/consensus.go b/pkg/raftstore/consensus/consensus.go index 145f53e4b..042de04d3 100644 --- a/pkg/raftstore/consensus/consensus.go +++ b/pkg/raftstore/consensus/consensus.go @@ -24,8 +24,9 @@ import ( type Consensus struct { logger *zap.Logger - nodeID uint64 - node raft.Node + locality string + nodeID uint64 + node raft.Node transport *rafthttp.Transport server *http.Server @@ -45,7 +46,7 @@ type Consensus struct { appliedIndex uint64 } -func NewConsensus(ctx context.Context, logger *zap.Logger, connectParams params.ConnectParameters, provider snapshotProvider, commitC chan<- EntryCommit) (*Consensus, error) { +func NewConsensus(ctx context.Context, logger *zap.Logger, locality string, connectParams params.ConnectParameters, provider snapshotProvider, commitC chan<- EntryCommit) (*Consensus, error) { storage, old, err := newStorage(ctx, logger.With(zap.String("component", "storage")), connectParams.DataDir, connectParams.NodeID, provider, connectParams.SnapshotCatchupEntries) if err != nil { return nil, stacktrace.Propagate(err, "failed to initialize storage") @@ -74,11 +75,11 @@ func NewConsensus(ctx context.Context, logger *zap.Logger, connectParams params. consensus := &Consensus{ logger: logging.WithValuesFromContext(ctx, logger), - nodeID: connectParams.NodeID, - node: node, - - storage: storage, - commitC: commitC, + nodeID: connectParams.NodeID, + node: node, + locality: locality, + storage: storage, + commitC: commitC, shutdownTimeout: 2 * connectParams.ElectionInterval(), serverErrC: make(chan error, 1), diff --git a/pkg/raftstore/consensus/proposal.go b/pkg/raftstore/consensus/proposal.go index 7e86f124e..eb002ced6 100644 --- a/pkg/raftstore/consensus/proposal.go +++ b/pkg/raftstore/consensus/proposal.go @@ -18,6 +18,7 @@ type EntryCommit struct { type Proposal struct { ID string `json:"id"` + Locality string `json:"locality"` NodeID uint64 `json:"node_id"` Timestamp time.Time `json:"timestamp"` RequestType RequestType `json:"request_type"` @@ -34,6 +35,7 @@ func (c *Consensus) newProposal(ctx context.Context, requestType RequestType, va return Proposal{ ID: uuid.NewString(), + Locality: c.locality, NodeID: c.nodeID, Timestamp: timestamp.UTC(), RequestType: requestType, diff --git a/pkg/raftstore/store.go b/pkg/raftstore/store.go index b01224d2a..83ef22b33 100644 --- a/pkg/raftstore/store.go +++ b/pkg/raftstore/store.go @@ -3,6 +3,7 @@ package raftstore import ( "context" + "github.com/interuss/dss/pkg/locality" "github.com/interuss/dss/pkg/logging" "github.com/interuss/dss/pkg/raftstore/consensus" raftparams "github.com/interuss/dss/pkg/raftstore/params" @@ -40,7 +41,7 @@ type Store[R any] struct { done chan struct{} } -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) { +func Init[R any](ctx context.Context, logger *zap.Logger, locality string, params raftparams.ConnectParameters, r RaftRepo[R], registry map[string]store.OperationHandler[R]) (*Store[R], error) { ctx, cancel := context.WithCancel(ctx) store := &Store[R]{ @@ -56,7 +57,7 @@ func Init[R any](ctx context.Context, logger *zap.Logger, params raftparams.Conn store.processCommits(ctx, commitC) }() - consensusInstance, err := consensus.NewConsensus(ctx, logger, params, r.GetSnapshot, commitC) + consensusInstance, err := consensus.NewConsensus(ctx, logger, locality, params, r.GetSnapshot, commitC) if err != nil { return nil, stacktrace.Propagate(err, "failed to initialize consensus") } @@ -115,6 +116,7 @@ func (s *Store[R]) processCommits(ctx context.Context, commitCh <-chan consensus } proposalCtx := timestamp.WithRequestTimestamp(ctx, commit.Prop.Timestamp) + proposalCtx = locality.WithRequestLocality(proposalCtx, commit.Prop.Locality) 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 bc55a3dca..a5636989c 100644 --- a/pkg/rid/store/raftstore/store.go +++ b/pkg/rid/store/raftstore/store.go @@ -16,12 +16,13 @@ import ( // repo is a full implementation of rid.repos.Repository for Raft-based storage. type repo struct{} -func Init(ctx context.Context, logger *zap.Logger) (*raftstore.Store[repos.Repository], error) { +func Init(ctx context.Context, logger *zap.Logger, locality string) (*raftstore.Store[repos.Repository], error) { params, err := ridraftparams.GetConnectParameters() 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, &repo{}, actions.Registry) + + return raftstore.Init(ctx, logger.With(zap.String("service", "rid")), locality, params, &repo{}, actions.Registry) } func (r *repo) GetRepo() repos.Repository { return r } diff --git a/pkg/rid/store/store.go b/pkg/rid/store/store.go index 7f2dc6c20..d9a1235c2 100644 --- a/pkg/rid/store/store.go +++ b/pkg/rid/store/store.go @@ -18,13 +18,13 @@ import ( type Store = dssstore.Store[repos.Repository] // Init selects and initializes the rid store backend. -func Init(ctx context.Context, logger *zap.Logger, withCheckCron bool) (Store, error) { +func Init(ctx context.Context, logger *zap.Logger, withCheckCron bool, locality string) (Store, error) { storeType := params.GetStoreParameters().StoreType switch storeType { case params.SQLStoreType: return ridsqlstore.Init(ctx, logger, withCheckCron) case params.RaftStoreType: - return ridraftstore.Init(ctx, logger) + return ridraftstore.Init(ctx, logger, locality) case params.MemStoreType: return ridmemstore.Init(ctx, logger) default: diff --git a/pkg/scd/actions/registry.go b/pkg/scd/actions/registry.go index c8389cc26..3f1193ef0 100644 --- a/pkg/scd/actions/registry.go +++ b/pkg/scd/actions/registry.go @@ -6,5 +6,4 @@ import ( ) // Registry maps operation IDs to their handlers -// TODO: implement var Registry = map[string]dssstore.OperationHandler[repos.Repository]{} diff --git a/pkg/scd/actions/subscription.go b/pkg/scd/actions/subscription.go new file mode 100644 index 000000000..4b76d617b --- /dev/null +++ b/pkg/scd/actions/subscription.go @@ -0,0 +1,182 @@ +package actions + +import ( + "context" + + restapi "github.com/interuss/dss/pkg/api/scdv1" + dsserr "github.com/interuss/dss/pkg/errors" + dssmodels "github.com/interuss/dss/pkg/models" + scdmodels "github.com/interuss/dss/pkg/scd/models" + "github.com/interuss/dss/pkg/scd/repos" + dssstore "github.com/interuss/dss/pkg/store" + "github.com/interuss/dss/pkg/timestamp" + "github.com/interuss/stacktrace" +) + +func init() { + Registry[restapi.DeleteSubscriptionOperationID] = dssstore.OperationHandler[repos.Repository]{ + Encode: dssstore.EncodeJSON, + Decode: dssstore.DecodeJSON[*restapi.DeleteSubscriptionRequest], + Execute: ExecuteDeleteSubscription, + } + Registry[restapi.GetSubscriptionOperationID] = dssstore.OperationHandler[repos.Repository]{ + Encode: dssstore.EncodeJSON, + Decode: dssstore.DecodeJSON[*restapi.GetSubscriptionRequest], + Execute: ExecuteGetSubscription, + IsReadOnly: true, + } + Registry[restapi.QuerySubscriptionsOperationID] = dssstore.OperationHandler[repos.Repository]{ + Encode: dssstore.EncodeJSON, + Decode: dssstore.DecodeJSON[*restapi.QuerySubscriptionsRequest], + Execute: ExecuteQuerySubscriptions, + IsReadOnly: true, + } +} + +func ExecuteDeleteSubscription(ctx context.Context, repo repos.Repository, request dssstore.OperationRequest) (any, error) { + req, ok := request.(*restapi.DeleteSubscriptionRequest) + if !ok { + return nil, stacktrace.NewError("unexpected request type %T for operation %q", request, restapi.DeleteSubscriptionOperationID) + } + + id, err := dssmodels.IDFromString(string(req.Subscriptionid)) + if err != nil { + return nil, stacktrace.NewErrorWithCode(dsserr.BadRequest, "Invalid subscription ID: %s", req.Subscriptionid) + } + + // Check to make sure it's ok to delete this Subscription + old, err := repo.GetSubscription(ctx, id) + switch { + case err != nil: + return nil, stacktrace.Propagate(err, "Could not get Subscription from repo") + case old == nil: // Return a 404 here. + return nil, stacktrace.NewErrorWithCode(dsserr.NotFound, "Subscription %s not found", id.String()) + case old.Manager != dssmodels.Manager(*req.Auth.ClientID): + return nil, stacktrace.Propagate( + stacktrace.NewErrorWithCode(dsserr.PermissionDenied, "Subscription is owned by different client"), + "Subscription owned by %s, but %s attempted to delete", old.Manager, *req.Auth.ClientID) + case old.Version != scdmodels.OVN(req.Version): + return nil, stacktrace.NewErrorWithCode(dsserr.VersionMismatch, "Subscription version %s is not current", scdmodels.OVN(req.Version)) + } + + // Get dependent Operations + dependentOps, err := repo.GetDependentOperationalIntents(ctx, id) + if err != nil { + return nil, stacktrace.Propagate(err, "Could not find dependent Operations") + } + if len(dependentOps) > 0 { + return nil, stacktrace.Propagate( + stacktrace.NewErrorWithCode(dsserr.BadRequest, "Subscriptions with dependent Operations may not be removed"), + "Subscription had %d dependent Operations", len(dependentOps)) + } + + // Delete Subscription in repo + err = repo.DeleteSubscription(ctx, id) + if err != nil { + return nil, stacktrace.Propagate(err, "Could not delete Subscription from repo") + } + + // Convert deleted Subscription to REST + p, err := old.ToRest(dependentOps) + if err != nil { + return nil, stacktrace.Propagate(err, "Error converting Subscription model to REST") + } + + return &restapi.DeleteSubscriptionResponse{Subscription: *p}, nil +} + +func ExecuteGetSubscription(ctx context.Context, repo repos.Repository, request dssstore.OperationRequest) (any, error) { + req, ok := request.(*restapi.GetSubscriptionRequest) + if !ok { + return nil, stacktrace.NewError("unexpected request type %T for operation %q", request, restapi.GetSubscriptionOperationID) + } + + // Retrieve Subscription ID + id, err := dssmodels.IDFromString(string(req.Subscriptionid)) + if err != nil { + return nil, stacktrace.NewErrorWithCode(dsserr.BadRequest, "Invalid ID format: `%s`", req.Subscriptionid) + } + + // Get Subscription from Store + sub, err := repo.GetSubscription(ctx, id) + if err != nil { + return nil, stacktrace.Propagate(err, "Could not get Subscription from repo") + } + if sub == nil { + return nil, stacktrace.NewErrorWithCode(dsserr.NotFound, "Subscription %s not found", id.String()) + } + + // Check if the client is authorized to view this Subscription + if dssmodels.Manager(*req.Auth.ClientID) != sub.Manager { + return nil, stacktrace.Propagate( + stacktrace.NewErrorWithCode(dsserr.PermissionDenied, "Subscription is owned by different client"), + "Subscription owned by %s, but %s attempted to view", sub.Manager, *req.Auth.ClientID) + } + + // Get dependent Operations + dependentOps, err := repo.GetDependentOperationalIntents(ctx, id) + if err != nil { + return nil, stacktrace.Propagate(err, "Could not find dependent Operations") + } + + // Convert Subscription to REST + p, err := sub.ToRest(dependentOps) + if err != nil { + return nil, stacktrace.Propagate(err, "Unable to convert Subscription to REST") + } + + // Return response to client + return &restapi.GetSubscriptionResponse{Subscription: *p}, nil +} + +func ExecuteQuerySubscriptions(ctx context.Context, repo repos.Repository, request dssstore.OperationRequest) (any, error) { + req, ok := request.(*restapi.QuerySubscriptionsRequest) + if !ok { + return nil, stacktrace.NewError("unexpected request type %T for operation %q", request, restapi.QuerySubscriptionsOperationID) + } + + // Retrieve the area of interest parameter + aoi := req.Body.AreaOfInterest + if aoi == nil { + return nil, stacktrace.NewErrorWithCode(dsserr.BadRequest, "Missing area_of_interest") + } + + // Parse area of interest to common Volume4D + vol4, err := scdmodels.Volume4DFromSCDRest(aoi) + if err != nil { + return nil, stacktrace.PropagateWithCode(err, dsserr.BadRequest, "Failed to convert to internal geometry model") + } + + // Perform search query on Store + subs, err := repo.SearchSubscriptions(ctx, vol4) + if err != nil { + return nil, stacktrace.Propagate(err, "Error searching Subscriptions in repo") + } + + nowMarker := timestamp.MustGetRequestTimestamp(ctx) + + // Return response to client + response := &restapi.QuerySubscriptionsResponse{ + Subscriptions: make([]restapi.Subscription, 0), + } + for _, sub := range subs { + // Do not return subscriptions which are expired. + // This implementation decision is described and motivated in https://github.com/interuss/tsc/pull/12. + isExpired := sub.EndTime.Before(nowMarker) + if !isExpired && sub.Manager == dssmodels.Manager(*req.Auth.ClientID) { + // Get dependent Operations + dependentOps, err := repo.GetDependentOperationalIntents(ctx, sub.ID) + if err != nil { + return nil, stacktrace.Propagate(err, "Could not find dependent Operations") + } + + p, err := sub.ToRest(dependentOps) + if err != nil { + return nil, stacktrace.Propagate(err, "Error converting Subscription model to REST") + } + response.Subscriptions = append(response.Subscriptions, *p) + } + } + + return response, nil +} diff --git a/pkg/scd/store/raftstore/store.go b/pkg/scd/store/raftstore/store.go index 799d6f1dd..918499c25 100644 --- a/pkg/scd/store/raftstore/store.go +++ b/pkg/scd/store/raftstore/store.go @@ -3,37 +3,70 @@ package raftstore import ( "context" - dsserr "github.com/interuss/dss/pkg/errors" + "github.com/interuss/dss/pkg/memstore" "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" + scdmemstore "github.com/interuss/dss/pkg/scd/store/memstore" scdraftparams "github.com/interuss/dss/pkg/scd/store/raftstore/params" "github.com/interuss/stacktrace" "go.uber.org/zap" ) // repo is a full implementation of scd.repos.Repository for Raft-based storage. -type repo struct{} +type repo struct { + consensus *consensus.Consensus + memStore *memstore.Store[repos.Repository] + memRepo repos.Repository +} -func Init(ctx context.Context, logger *zap.Logger) (*raftstore.Store[repos.Repository], error) { +func Init(ctx context.Context, logger *zap.Logger, locality string) (*raftstore.Store[repos.Repository], error) { params, err := scdraftparams.GetConnectParameters() 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, &repo{}, actions.Registry) + + memStore, err := scdmemstore.Init(ctx, logger) + if err != nil { + return nil, stacktrace.Propagate(err, "failed to initialize scd memstore") + } + + r := &repo{memStore: memStore, memRepo: memStore.GetRepo()} + store, err := raftstore.Init(ctx, logger.With(zap.String("service", "scd")), locality, params, r, actions.Registry) + if err != nil { + return nil, stacktrace.Propagate(err, "failed to initialize scd raftstore") + } + + r.consensus = store.Consensus + + return store, nil } func (r *repo) GetRepo() repos.Repository { return r } func (r *repo) GetSnapshot() ([]byte, error) { - return nil, stacktrace.NewErrorWithCode(dsserr.NotImplemented, "not implemented yet") + return r.memStore.GetSnapshot() } -func (r *repo) RestoreFromSnapshot([]byte) error { - return stacktrace.NewErrorWithCode(dsserr.NotImplemented, "not implemented yet") +func (r *repo) RestoreFromSnapshot(data []byte) error { + return r.memStore.RestoreFromSnapshot(data) } -func (r *repo) Apply(_ context.Context, _ consensus.Proposal) (any, error) { - return nil, stacktrace.NewErrorWithCode(dsserr.NotImplemented, "not implemented yet") +func (r *repo) Apply(ctx context.Context, proposal consensus.Proposal) (any, error) { + switch proposal.RequestType { + + default: + handler, ok := actions.Registry[string(proposal.RequestType)] + if !ok { + return nil, stacktrace.NewError("unrecognized request type: %s", proposal.RequestType) + } + + request, err := handler.Decode(proposal.Value) + if err != nil { + return nil, stacktrace.Propagate(err, "failed to decode %s payload", proposal.RequestType) + } + + return handler.Execute(ctx, r.memRepo, request) + } } diff --git a/pkg/scd/store/store.go b/pkg/scd/store/store.go index 0cdc48b93..6a6b70665 100644 --- a/pkg/scd/store/store.go +++ b/pkg/scd/store/store.go @@ -18,13 +18,13 @@ import ( type Store = dssstore.Store[repos.Repository] // Init selects and initializes the scd store backend. -func Init(ctx context.Context, logger *zap.Logger, withCheckCron bool) (Store, error) { +func Init(ctx context.Context, logger *zap.Logger, withCheckCron bool, locality string) (Store, error) { storeType := params.GetStoreParameters().StoreType switch storeType { case params.SQLStoreType: return scdsqlstore.Init(ctx, logger, withCheckCron) case params.RaftStoreType: - return scdraftstore.Init(ctx, logger) + return scdraftstore.Init(ctx, logger, locality) case params.MemStoreType: return scdmemstore.Init(ctx, logger) default: diff --git a/pkg/scd/subscriptions_handler.go b/pkg/scd/subscriptions_handler.go index 7619b03a0..fbe0aa1fb 100644 --- a/pkg/scd/subscriptions_handler.go +++ b/pkg/scd/subscriptions_handler.go @@ -2,7 +2,6 @@ package scd import ( "context" - "time" "github.com/golang/geo/s2" "github.com/interuss/dss/pkg/api" @@ -290,7 +289,7 @@ func (a *Server) GetSubscription(ctx context.Context, req *restapi.GetSubscripti ) restapi.GetSubscriptionResponseSet { // Retrieve Subscription ID - id, err := dssmodels.IDFromString(string(req.Subscriptionid)) + _, err := dssmodels.IDFromString(string(req.Subscriptionid)) if err != nil { return restapi.GetSubscriptionResponseSet{Response400: &restapi.ErrorResponse{ Message: dsserr.Handle(ctx, stacktrace.NewErrorWithCode(dsserr.BadRequest, "Invalid ID format: `%s`", req.Subscriptionid))}} @@ -302,45 +301,7 @@ func (a *Server) GetSubscription(ctx context.Context, req *restapi.GetSubscripti Message: dsserr.Handle(ctx, stacktrace.NewErrorWithCode(dsserr.PermissionDenied, "Missing owner"))}} } - var response *restapi.GetSubscriptionResponse - action := func(ctx context.Context, r repos.Repository) (err error) { - // Get Subscription from Store - sub, err := r.GetSubscription(ctx, id) - if err != nil { - return stacktrace.Propagate(err, "Could not get Subscription from repo") - } - if sub == nil { - return stacktrace.NewErrorWithCode(dsserr.NotFound, "Subscription %s not found", id.String()) - } - - // Check if the client is authorized to view this Subscription - if dssmodels.Manager(*req.Auth.ClientID) != sub.Manager { - return stacktrace.Propagate( - stacktrace.NewErrorWithCode(dsserr.PermissionDenied, "Subscription is owned by different client"), - "Subscription owned by %s, but %s attempted to view", sub.Manager, *req.Auth.ClientID) - } - - // Get dependent Operations - dependentOps, err := r.GetDependentOperationalIntents(ctx, id) - if err != nil { - return stacktrace.Propagate(err, "Could not find dependent Operations") - } - - // Convert Subscription to REST - p, err := sub.ToRest(dependentOps) - if err != nil { - return stacktrace.Propagate(err, "Unable to convert Subscription to REST") - } - - // Return response to client - response = &restapi.GetSubscriptionResponse{ - Subscription: *p, - } - - return nil - } - - _, err = a.Store.Transact(ctx, dssstore.NewFuncOperation(action)) + response, err := dssstore.TransactWithResult[repos.Repository, *restapi.GetSubscriptionResponse](ctx, a.Store, req) if err != nil { err = stacktrace.Propagate(err, "Could not get subscription") errResp := &restapi.ErrorResponse{Message: dsserr.Handle(ctx, err)} @@ -363,7 +324,6 @@ func (a *Server) GetSubscription(ctx context.Context, req *restapi.GetSubscripti // QuerySubscriptions queries existing subscriptions in the given bounds. func (a *Server) QuerySubscriptions(ctx context.Context, req *restapi.QuerySubscriptionsRequest, ) restapi.QuerySubscriptionsResponseSet { - nowMarker := time.Now() if req.BodyParseError != nil { return restapi.QuerySubscriptionsResponseSet{Response400: &restapi.ErrorResponse{ @@ -378,7 +338,7 @@ func (a *Server) QuerySubscriptions(ctx context.Context, req *restapi.QuerySubsc } // Parse area of interest to common Volume4D - vol4, err := scdmodels.Volume4DFromSCDRest(aoi) + _, err := scdmodels.Volume4DFromSCDRest(aoi) if err != nil { return restapi.QuerySubscriptionsResponseSet{Response400: &restapi.ErrorResponse{ Message: dsserr.Handle(ctx, stacktrace.PropagateWithCode(err, dsserr.BadRequest, "Failed to convert to internal geometry model"))}} @@ -390,41 +350,7 @@ func (a *Server) QuerySubscriptions(ctx context.Context, req *restapi.QuerySubsc Message: dsserr.Handle(ctx, stacktrace.NewErrorWithCode(dsserr.PermissionDenied, "Missing owner"))}} } - var response *restapi.QuerySubscriptionsResponse - action := func(ctx context.Context, r repos.Repository) (err error) { - // Perform search query on Store - subs, err := r.SearchSubscriptions(ctx, vol4) - if err != nil { - return stacktrace.Propagate(err, "Error searching Subscriptions in repo") - } - - // Return response to client - response = &restapi.QuerySubscriptionsResponse{ - Subscriptions: make([]restapi.Subscription, 0), - } - for _, sub := range subs { - // Do not return subscriptions which are expired. - // This implementation decision is described and motivated in https://github.com/interuss/tsc/pull/12. - isExpired := sub.EndTime.Before(nowMarker) - if !isExpired && sub.Manager == dssmodels.Manager(*req.Auth.ClientID) { - // Get dependent Operations - dependentOps, err := r.GetDependentOperationalIntents(ctx, sub.ID) - if err != nil { - return stacktrace.Propagate(err, "Could not find dependent Operations") - } - - p, err := sub.ToRest(dependentOps) - if err != nil { - return stacktrace.Propagate(err, "Error converting Subscription model to REST") - } - response.Subscriptions = append(response.Subscriptions, *p) - } - } - - return nil - } - - _, err = a.Store.Transact(ctx, dssstore.NewFuncOperation(action)) + response, err := dssstore.TransactWithResult[repos.Repository, *restapi.QuerySubscriptionsResponse](ctx, a.Store, req) if err != nil { errResp := &restapi.ErrorResponse{Message: dsserr.Handle(ctx, err)} @@ -450,7 +376,7 @@ func (a *Server) DeleteSubscription(ctx context.Context, req *restapi.DeleteSubs ) restapi.DeleteSubscriptionResponseSet { // Retrieve Subscription ID - id, err := dssmodels.IDFromString(string(req.Subscriptionid)) + _, err := dssmodels.IDFromString(string(req.Subscriptionid)) if err != nil { return restapi.DeleteSubscriptionResponseSet{Response400: &restapi.ErrorResponse{ Message: dsserr.Handle(ctx, stacktrace.NewErrorWithCode(dsserr.BadRequest, "Invalid ID format"))}} @@ -469,55 +395,7 @@ func (a *Server) DeleteSubscription(ctx context.Context, req *restapi.DeleteSubs Message: dsserr.Handle(ctx, stacktrace.NewErrorWithCode(dsserr.PermissionDenied, "Missing owner"))}} } - var response *restapi.DeleteSubscriptionResponse - action := func(ctx context.Context, r repos.Repository) (err error) { - // Check to make sure it's ok to delete this Subscription - old, err := r.GetSubscription(ctx, id) - switch { - case err != nil: - return stacktrace.Propagate(err, "Could not get Subscription from repo") - case old == nil: // Return a 404 here. - return stacktrace.NewErrorWithCode(dsserr.NotFound, "Subscription %s not found", id.String()) - case old.Manager != dssmodels.Manager(*req.Auth.ClientID): - return stacktrace.Propagate( - stacktrace.NewErrorWithCode(dsserr.PermissionDenied, "Subscription is owned by different client"), - "Subscription owned by %s, but %s attempted to delete", old.Manager, *req.Auth.ClientID) - case old.Version != version: - return stacktrace.NewErrorWithCode(dsserr.VersionMismatch, "Subscription version %s is not current", version) - } - - // Get dependent Operations - dependentOps, err := r.GetDependentOperationalIntents(ctx, id) - if err != nil { - return stacktrace.Propagate(err, "Could not find dependent Operations") - } - if len(dependentOps) > 0 { - return stacktrace.Propagate( - stacktrace.NewErrorWithCode(dsserr.BadRequest, "Subscriptions with dependent Operations may not be removed"), - "Subscription had %d dependent Operations", len(dependentOps)) - } - - // Delete Subscription in repo - err = r.DeleteSubscription(ctx, id) - if err != nil { - return stacktrace.Propagate(err, "Could not delete Subscription from repo") - } - - // Convert deleted Subscription to REST - p, err := old.ToRest(dependentOps) - if err != nil { - return stacktrace.Propagate(err, "Error converting Subscription model to REST") - } - - // Create response for client - response = &restapi.DeleteSubscriptionResponse{ - Subscription: *p, - } - - return nil - } - - _, err = a.Store.Transact(ctx, dssstore.NewFuncOperation(action)) + response, err := dssstore.TransactWithResult[repos.Repository, *restapi.DeleteSubscriptionResponse](ctx, a.Store, req) if err != nil { err = stacktrace.Propagate(err, "Could not delete subscription") errResp := &restapi.ErrorResponse{Message: dsserr.Handle(ctx, err)}