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/availability.go b/pkg/scd/actions/availability.go new file mode 100644 index 000000000..2942d9d9e --- /dev/null +++ b/pkg/scd/actions/availability.go @@ -0,0 +1,109 @@ +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/stacktrace" + "github.com/jackc/pgx/v5" +) + +func init() { + Registry[restapi.GetUssAvailabilityOperationID] = dssstore.OperationHandler[repos.Repository]{ + Encode: dssstore.EncodeJSON, + Decode: dssstore.DecodeJSON[*restapi.GetUssAvailabilityRequest], + Execute: ExecuteGetUssAvailability, + IsReadOnly: true, + } + Registry[restapi.SetUssAvailabilityOperationID] = dssstore.OperationHandler[repos.Repository]{ + Encode: dssstore.EncodeJSON, + Decode: dssstore.DecodeJSON[*restapi.SetUssAvailabilityRequest], + Execute: ExecuteSetUssAvailability, + } +} + +func GetDefaultAvailabilityResponse(id dssmodels.Manager) *restapi.UssAvailabilityStatusResponse { + return &restapi.UssAvailabilityStatusResponse{ + Status: restapi.UssAvailabilityStatus{ + Availability: restapi.UssAvailabilityState_Unknown, + Uss: id.String()}, + Version: "", + } +} + +func ExecuteGetUssAvailability(ctx context.Context, repo repos.Repository, request dssstore.OperationRequest) (any, error) { + req, ok := request.(*restapi.GetUssAvailabilityRequest) + if !ok { + return nil, stacktrace.NewError("unexpected request type %T for operation %q", request, restapi.GetUssAvailabilityOperationID) + } + + id := dssmodels.ManagerFromString(req.UssId) + + // Get USS availability from Store + ussa, err := repo.GetUssAvailability(ctx, id) + if err != nil && err != pgx.ErrNoRows { + return nil, stacktrace.Propagate(err, "Could not get USS availability from repo") + } + if ussa == nil { + // Return default availability status "Unknown" + return GetDefaultAvailabilityResponse(id), nil + } + + return &restapi.UssAvailabilityStatusResponse{ + Status: *ussa.ToRest(), + Version: ussa.Version.String(), + }, nil +} + +func ExecuteSetUssAvailability(ctx context.Context, repo repos.Repository, request dssstore.OperationRequest) (any, error) { + req, ok := request.(*restapi.SetUssAvailabilityRequest) + if !ok { + return nil, stacktrace.NewError("unexpected request type %T for operation %q", request, restapi.SetUssAvailabilityOperationID) + } + + // Retrieve USS availability status from request params + availability, err := scdmodels.UssAvailabilityStateFromRest(req.Body.Availability) + if err != nil { + return nil, stacktrace.PropagateWithCode(err, dsserr.BadRequest, "Invalid availability state") + } + id := dssmodels.ManagerFromString(req.UssId) + version := scdmodels.OVN(req.Body.OldVersion) + ussareq := &scdmodels.UssAvailabilityStatus{ + Uss: id, + Availability: availability, + } + + old, err := repo.GetUssAvailability(ctx, id) + if err != nil && err != pgx.ErrNoRows { + return nil, stacktrace.Propagate(err, "Could not get USS availability from repo") + } + switch { + case old == nil && !version.Empty(): + // The user wants set a new availability status but it already exists. + return nil, stacktrace.NewErrorWithCode(dsserr.AlreadyExists, "availability for USS %s already exists", id.String()) + case old != nil && old.Version != version: + // The user wants to update an availability status but the version doesn't match. + return nil, stacktrace.Propagate( + stacktrace.NewErrorWithCode(dsserr.VersionMismatch, "USS availability version %s is not current", version), + "Current version is %s but client specified version %s", old.Version, version) + } + + // Upsert the USS availability + ussa, err := repo.UpsertUssAvailability(ctx, ussareq) + if err != nil { + return nil, stacktrace.Propagate(err, "Could not upsert USS Availability into repo") + } + if ussa == nil { + return nil, stacktrace.NewError("UpsertUssAvailability returned no USS availability for ID: %s", id) + } + + return &restapi.UssAvailabilityStatusResponse{ + Status: *ussa.ToRest(), + Version: ussa.Version.String(), + }, nil +} diff --git a/pkg/scd/actions/constraint.go b/pkg/scd/actions/constraint.go new file mode 100644 index 000000000..6fff92478 --- /dev/null +++ b/pkg/scd/actions/constraint.go @@ -0,0 +1,181 @@ +package actions + +import ( + "context" + + "github.com/golang/geo/s2" + 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/stacktrace" + "github.com/jackc/pgx/v5" +) + +func init() { + Registry[restapi.DeleteConstraintReferenceOperationID] = dssstore.OperationHandler[repos.Repository]{ + Encode: dssstore.EncodeJSON, + Decode: dssstore.DecodeJSON[*restapi.DeleteConstraintReferenceRequest], + Execute: ExecuteDeleteConstraint, + } + Registry[restapi.GetConstraintReferenceOperationID] = dssstore.OperationHandler[repos.Repository]{ + Encode: dssstore.EncodeJSON, + Decode: dssstore.DecodeJSON[*restapi.GetConstraintReferenceRequest], + Execute: ExecuteGetConstraint, + IsReadOnly: true, + } + Registry[restapi.QueryConstraintReferencesOperationID] = dssstore.OperationHandler[repos.Repository]{ + Encode: dssstore.EncodeJSON, + Decode: dssstore.DecodeJSON[*restapi.QueryConstraintReferencesRequest], + Execute: ExecuteQueryConstraintReferences, + IsReadOnly: true, + } +} + +func ExecuteGetConstraint(ctx context.Context, repo repos.Repository, request dssstore.OperationRequest) (any, error) { + req, ok := request.(*restapi.GetConstraintReferenceRequest) + if !ok { + return nil, stacktrace.NewError("unexpected request type %T for operation %q", request, restapi.GetConstraintReferenceOperationID) + } + + id, err := dssmodels.IDFromString(string(req.Entityid)) + if err != nil { + return nil, stacktrace.NewErrorWithCode(dsserr.BadRequest, "Invalid ID format: `%s`", req.Entityid) + } + + constraint, err := repo.GetConstraint(ctx, id) + switch { + case err == pgx.ErrNoRows: + return nil, stacktrace.NewErrorWithCode(dsserr.NotFound, "Constraint %s not found", id.String()) + case err != nil: + return nil, stacktrace.Propagate(err, "Unable to get Constraint from repo") + } + + if constraint.Manager != dssmodels.Manager(*req.Auth.ClientID) { + constraint.OVN = scdmodels.NoOvnPhrase + } + + // Return response to client + return &restapi.GetConstraintReferenceResponse{ + ConstraintReference: *constraint.ToRest(), + }, nil +} + +func ExecuteQueryConstraintReferences(ctx context.Context, repo repos.Repository, request dssstore.OperationRequest) (any, error) { + req, ok := request.(*restapi.QueryConstraintReferencesRequest) + if !ok { + return nil, stacktrace.NewError("unexpected request type %T for operation %q", request, restapi.QueryConstraintReferencesOperationID) + } + + // 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 + constraints, err := repo.SearchConstraints(ctx, vol4) + if err != nil { + return nil, err + } + + // Create response for client + response := &restapi.QueryConstraintReferencesResponse{ + ConstraintReferences: make([]restapi.ConstraintReference, 0, len(constraints)), + } + for _, constraint := range constraints { + p := constraint.ToRest() + if constraint.Manager != dssmodels.Manager(*req.Auth.ClientID) { + noOvnPhrase := restapi.EntityOVN(scdmodels.NoOvnPhrase) + p.Ovn = &noOvnPhrase + } + response.ConstraintReferences = append(response.ConstraintReferences, *p) + } + + return response, nil +} + +func ExecuteDeleteConstraint(ctx context.Context, repo repos.Repository, request dssstore.OperationRequest) (any, error) { + req, ok := request.(*restapi.DeleteConstraintReferenceRequest) + if !ok { + return nil, stacktrace.NewError("unexpected request type %T for operation %q", request, restapi.DeleteConstraintReferenceOperationID) + } + + // Retrieve Constraint ID + id, err := dssmodels.IDFromString(string(req.Entityid)) + if err != nil { + return nil, stacktrace.NewErrorWithCode(dsserr.BadRequest, "Invalid ID format: `%s`", req.Entityid) + } + + // Make sure deletion request is valid + old, err := repo.GetConstraint(ctx, id) + switch { + case err == pgx.ErrNoRows: + return nil, stacktrace.NewErrorWithCode(dsserr.NotFound, "Constraint %s not found", id.String()) + case err != nil: + return nil, stacktrace.Propagate(err, "Unable to get Constraint from repo") + case old.Manager != dssmodels.Manager(*req.Auth.ClientID): + return nil, stacktrace.NewErrorWithCode(dsserr.PermissionDenied, + "Constraint owned by %s, but %s attempted to delete", old.Manager, *req.Auth.ClientID) + case old.OVN != scdmodels.OVN(req.Ovn): + return nil, stacktrace.NewErrorWithCode(dsserr.VersionMismatch, + "Current version is %s but client specified version %s", old.OVN, scdmodels.OVN(req.Ovn)) + } + + // Delete Constraint in repo + err = repo.DeleteConstraint(ctx, id) + if err != nil { + return nil, stacktrace.Propagate(err, "Unable to delete Constraint from repo") + } + + // Find the Subscriptions interested in Constraints and increment their + // notification indices. + subs, err := repo.IncrementNotificationIndicesForConstraints(ctx, &dssmodels.Volume4D{ + StartTime: old.StartTime, + EndTime: old.EndTime, + SpatialVolume: &dssmodels.Volume3D{ + AltitudeHi: old.AltitudeUpper, + AltitudeLo: old.AltitudeLower, + Footprint: dssmodels.GeometryFunc(func() (s2.CellUnion, error) { + return old.Cells, nil + }), + }}) + if err != nil { + return nil, stacktrace.Propagate(err, "Unable to increment notification indices") + } + + // Return response to client + return &restapi.ChangeConstraintReferenceResponse{ + ConstraintReference: *old.ToRest(), + Subscribers: makeSubscribersToNotify(subs), + }, nil +} + +func makeSubscribersToNotify(subscriptions []*scdmodels.Subscription) []restapi.SubscriberToNotify { + result := []restapi.SubscriberToNotify{} + + subscriptionsByURL := map[string][]restapi.SubscriptionState{} + for _, sub := range subscriptions { + subState := restapi.SubscriptionState{ + SubscriptionId: restapi.SubscriptionID(sub.ID.String()), + NotificationIndex: restapi.SubscriptionNotificationIndex(sub.NotificationIndex), + } + subscriptionsByURL[sub.USSBaseURL] = append(subscriptionsByURL[sub.USSBaseURL], subState) + } + for url, states := range subscriptionsByURL { + result = append(result, restapi.SubscriberToNotify{ + UssBaseUrl: restapi.SubscriptionUssBaseURL(url), + Subscriptions: states, + }) + } + + return result +} 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..81b842c4c --- /dev/null +++ b/pkg/scd/actions/subscription.go @@ -0,0 +1,402 @@ +package actions + +import ( + "context" + + "github.com/golang/geo/s2" + restapi "github.com/interuss/dss/pkg/api/scdv1" + dsserr "github.com/interuss/dss/pkg/errors" + "github.com/interuss/dss/pkg/geo" + 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.CreateSubscriptionOperationID] = dssstore.OperationHandler[repos.Repository]{ + Encode: dssstore.EncodeJSON, + Decode: dssstore.DecodeJSON[*restapi.CreateSubscriptionRequest], + Execute: ExecutePutSubscription, + } + Registry[restapi.UpdateSubscriptionOperationID] = dssstore.OperationHandler[repos.Repository]{ + Encode: dssstore.EncodeJSON, + Decode: dssstore.DecodeJSON[*restapi.UpdateSubscriptionRequest], + Execute: ExecutePutSubscription, + } + 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 ExecutePutSubscription(ctx context.Context, repo repos.Repository, request dssstore.OperationRequest) (any, error) { + var ( + manager string + subscriptionid restapi.SubscriptionID + version string + params *restapi.PutSubscriptionParameters + ) + + switch req := request.(type) { + case *restapi.CreateSubscriptionRequest: + manager, subscriptionid, params = *req.Auth.ClientID, req.Subscriptionid, req.Body + case *restapi.UpdateSubscriptionRequest: + manager, subscriptionid, version, params = *req.Auth.ClientID, req.Subscriptionid, req.Version, req.Body + default: + return nil, stacktrace.NewError("unexpected request type %T for operation %q", request, restapi.CreateSubscriptionOperationID) + } + + // Retrieve Subscription ID + id, err := dssmodels.IDFromString(string(subscriptionid)) + if err != nil { + return nil, stacktrace.NewErrorWithCode(dsserr.BadRequest, "Invalid ID format: `%s`", subscriptionid) + } + + // Parse extents + // If end time is not specified, the value will be chosen automatically by the DSS. + // If start time is not specified, it will default to the time the request is processed. + extents, err := scdmodels.Volume4DFromSCDRest(¶ms.Extents) + if err != nil { + return nil, stacktrace.PropagateWithCode(err, dsserr.BadRequest, "Unable to parse extents") + } + + // Construct requested Subscription model + cells, err := extents.CalculateSpatialCovering() + switch err { + case nil, geo.ErrMissingSpatialVolume, geo.ErrMissingFootprint: + // We may be able to fill these values from a previous Subscription or via defaults. + default: + return nil, stacktrace.PropagateWithCode(err, dsserr.BadRequest, "Invalid area") + } + + subreq := &scdmodels.Subscription{ + ID: id, + Manager: dssmodels.Manager(manager), + Version: scdmodels.OVN(version), + + StartTime: extents.StartTime, + EndTime: extents.EndTime, + AltitudeLo: extents.SpatialVolume.AltitudeLo, + AltitudeHi: extents.SpatialVolume.AltitudeHi, + Cells: cells, + + USSBaseURL: string(params.UssBaseUrl), + } + if params.NotifyForOperationalIntents != nil { + subreq.NotifyForOperationalIntents = *params.NotifyForOperationalIntents + } + if params.NotifyForConstraints != nil { + subreq.NotifyForConstraints = *params.NotifyForConstraints + } + + // Validate requested Subscription + if !subreq.NotifyForOperationalIntents && !subreq.NotifyForConstraints { + return nil, stacktrace.NewErrorWithCode(dsserr.BadRequest, "No notification triggers requested for Subscription") + } + + // TODO: Check scopes to verify requested information (op intents or constraints) may be requested + + // Check existing Subscription (if any) + old, err := repo.GetSubscription(ctx, subreq.ID) + if err != nil { + return nil, stacktrace.Propagate(err, "Could not get Subscription from repo") + } + + // Validate and perhaps correct StartTime and EndTime. + if err := subreq.AdjustTimeRange(timestamp.MustGetRequestTimestamp(ctx), old); err != nil { + return nil, stacktrace.Propagate(err, "Error adjusting time range of Subscription") + } + + var dependentOpIds []dssmodels.ID + + if old == nil { + // There is no previous Subscription (this is a creation attempt) + if subreq.Version.String() != "" { + // The user wants to update an existing Subscription, but one wasn't found. + return nil, stacktrace.NewErrorWithCode(dsserr.NotFound, "Subscription %s not found", subreq.ID.String()) + } + } else { + // There is a previous Subscription (this is an update attempt) + switch { + case subreq.Version.String() == "": + // The user wants to create a new Subscription but it already exists. + return nil, stacktrace.NewErrorWithCode(dsserr.AlreadyExists, "Subscription %s already exists", subreq.ID.String()) + case subreq.Version.String() != old.Version.String(): + // The user wants to update a Subscription but the version doesn't match. + return nil, stacktrace.Propagate( + stacktrace.NewErrorWithCode(dsserr.VersionMismatch, "Subscription version %s is not current", subreq.Version), + "Current version is %s but client specified version %s", old.Version, subreq.Version) + case old.Manager != subreq.Manager: + return nil, stacktrace.Propagate( + stacktrace.NewErrorWithCode(dsserr.PermissionDenied, "Subscription is owned by different client"), + "Subscription owned by %s, but %s attempted to modify", old.Manager, subreq.Manager) + } + + subreq.NotificationIndex = old.NotificationIndex + + // Validate Subscription against DependentOperations + dependentOpIds, err = repo.GetDependentOperationalIntents(ctx, subreq.ID) + if err != nil { + return nil, stacktrace.Propagate(err, "Could not find dependent Operation Ids") + } + + operations, err := GetOperations(ctx, repo, dependentOpIds) + if err != nil { + return nil, stacktrace.Propagate(err, "Could not get all dependent Operations") + } + if err := subreq.ValidateDependentOps(operations); err != nil { + // The provided subscription does not cover all its dependent operations + return nil, err + } + } + + // Store Subscription model + sub, err := repo.UpsertSubscription(ctx, subreq) + if err != nil { + return nil, stacktrace.Propagate(err, "Could not upsert Subscription into repo") + } + if sub == nil { + return nil, stacktrace.NewError("UpsertSubscription returned no Subscription for ID: %s", id) + } + + // Convert Subscription to REST + p, err := sub.ToRest(dependentOpIds) + if err != nil { + return nil, stacktrace.Propagate(err, "Could not convert Subscription to REST model") + } + result := &restapi.PutSubscriptionResponse{ + Subscription: *p, + } + + if sub.NotifyForOperationalIntents { + // Find relevant Operations + var relevantOperations []*scdmodels.OperationalIntent + if len(sub.Cells) > 0 { + ops, err := repo.SearchOperationalIntents(ctx, &dssmodels.Volume4D{ + StartTime: sub.StartTime, + EndTime: sub.EndTime, + SpatialVolume: &dssmodels.Volume3D{ + AltitudeLo: sub.AltitudeLo, + AltitudeHi: sub.AltitudeHi, + Footprint: dssmodels.GeometryFunc(func() (s2.CellUnion, error) { + return sub.Cells, nil + }), + }, + }) + if err != nil { + return nil, stacktrace.Propagate(err, "Could not search Operations in repo") + } + relevantOperations = ops + } + // Attach Operations to response + opIntentRefs := make([]restapi.OperationalIntentReference, 0, len(relevantOperations)) + for _, op := range relevantOperations { + if op.Manager != dssmodels.Manager(manager) { + op.OVN = scdmodels.NoOvnPhrase + } + + opIntentRefs = append(opIntentRefs, *op.ToRest()) + } + result.OperationalIntentReferences = &opIntentRefs + } + + if sub.NotifyForConstraints { + // Query relevant Constraints + constraints, err := repo.SearchConstraints(ctx, extents) + if err != nil { + return nil, stacktrace.Propagate(err, "Could not search Constraints in repo") + } + + // Attach Constraints to response + constraintRefs := make([]restapi.ConstraintReference, 0, len(constraints)) + for _, constraint := range constraints { + p := constraint.ToRest() + if constraint.Manager != dssmodels.Manager(manager) { + noOvnPhrase := restapi.EntityOVN(scdmodels.NoOvnPhrase) + p.Ovn = &noOvnPhrase + } + + constraintRefs = append(constraintRefs, *p) + } + result.ConstraintReferences = &constraintRefs + } + + return result, nil +} + +// GetOperations gets operations by given ids +func GetOperations(ctx context.Context, r repos.Repository, opIDs []dssmodels.ID) ([]*scdmodels.OperationalIntent, error) { + var res []*scdmodels.OperationalIntent + for _, opID := range opIDs { + operation, err := r.GetOperationalIntent(ctx, opID) + if err != nil { + return nil, stacktrace.Propagate(err, "Could not retrieve dependent Operation %s", opID) + } + res = append(res, operation) + } + return res, nil +} + +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/constraints_handler.go b/pkg/scd/constraints_handler.go index 851ccb3f7..5550a6bd2 100644 --- a/pkg/scd/constraints_handler.go +++ b/pkg/scd/constraints_handler.go @@ -22,7 +22,7 @@ func (a *Server) DeleteConstraintReference(ctx context.Context, req *restapi.Del ) restapi.DeleteConstraintReferenceResponseSet { // Retrieve Constraint ID - id, err := dssmodels.IDFromString(string(req.Entityid)) + _, err := dssmodels.IDFromString(string(req.Entityid)) if err != nil { return restapi.DeleteConstraintReferenceResponseSet{Response400: &restapi.ErrorResponse{ Message: dsserr.Handle(ctx, stacktrace.NewErrorWithCode(dsserr.BadRequest, "Invalid ID format: `%s`", req.Entityid))}} @@ -41,55 +41,7 @@ func (a *Server) DeleteConstraintReference(ctx context.Context, req *restapi.Del Message: dsserr.Handle(ctx, stacktrace.NewErrorWithCode(dsserr.BadRequest, "Missing OVN for constraint to modify"))}} } - var response *restapi.ChangeConstraintReferenceResponse - action := func(ctx context.Context, r repos.Repository) (err error) { - // Make sure deletion request is valid - old, err := r.GetConstraint(ctx, id) - switch { - case err == pgx.ErrNoRows: - return stacktrace.NewErrorWithCode(dsserr.NotFound, "Constraint %s not found", id.String()) - case err != nil: - return stacktrace.Propagate(err, "Unable to get Constraint from repo") - case old.Manager != dssmodels.Manager(*req.Auth.ClientID): - return stacktrace.NewErrorWithCode(dsserr.PermissionDenied, - "Constraint owned by %s, but %s attempted to delete", old.Manager, *req.Auth.ClientID) - case old.OVN != ovn: - return stacktrace.NewErrorWithCode(dsserr.VersionMismatch, - "Current version is %s but client specified version %s", old.OVN, ovn) - } - - // Delete Constraint in repo - err = r.DeleteConstraint(ctx, id) - if err != nil { - return stacktrace.Propagate(err, "Unable to delete Constraint from repo") - } - - // Find the Subscriptions interested in Constraints and increment their - // notification indices. - subs, err := r.IncrementNotificationIndicesForConstraints(ctx, &dssmodels.Volume4D{ - StartTime: old.StartTime, - EndTime: old.EndTime, - SpatialVolume: &dssmodels.Volume3D{ - AltitudeHi: old.AltitudeUpper, - AltitudeLo: old.AltitudeLower, - Footprint: dssmodels.GeometryFunc(func() (s2.CellUnion, error) { - return old.Cells, nil - }), - }}) - if err != nil { - return stacktrace.Propagate(err, "Unable to increment notification indices") - } - - // Return response to client - response = &restapi.ChangeConstraintReferenceResponse{ - ConstraintReference: *old.ToRest(), - Subscribers: makeSubscribersToNotify(subs), - } - - return nil - } - - _, err = a.Store.Transact(ctx, dssstore.NewFuncOperation(action)) + response, err := dssstore.TransactWithResult[repos.Repository, *restapi.ChangeConstraintReferenceResponse](ctx, a.Store, req) if err != nil { err = stacktrace.Propagate(err, "Could not delete constraint") errResp := &restapi.ErrorResponse{Message: dsserr.Handle(ctx, err)} @@ -115,7 +67,7 @@ func (a *Server) DeleteConstraintReference(ctx context.Context, req *restapi.Del func (a *Server) GetConstraintReference(ctx context.Context, req *restapi.GetConstraintReferenceRequest, ) restapi.GetConstraintReferenceResponseSet { - id, err := dssmodels.IDFromString(string(req.Entityid)) + _, err := dssmodels.IDFromString(string(req.Entityid)) if err != nil { return restapi.GetConstraintReferenceResponseSet{Response400: &restapi.ErrorResponse{ Message: dsserr.Handle(ctx, stacktrace.NewErrorWithCode(dsserr.BadRequest, "Invalid ID format: `%s`", req.Entityid))}} @@ -126,29 +78,7 @@ func (a *Server) GetConstraintReference(ctx context.Context, req *restapi.GetCon Message: dsserr.Handle(ctx, stacktrace.NewErrorWithCode(dsserr.PermissionDenied, "Missing manager"))}} } - var response *restapi.GetConstraintReferenceResponse - action := func(ctx context.Context, r repos.Repository) (err error) { - constraint, err := r.GetConstraint(ctx, id) - switch { - case err == pgx.ErrNoRows: - return stacktrace.NewErrorWithCode(dsserr.NotFound, "Constraint %s not found", id.String()) - case err != nil: - return stacktrace.Propagate(err, "Unable to get Constraint from repo") - } - - if constraint.Manager != dssmodels.Manager(*req.Auth.ClientID) { - constraint.OVN = scdmodels.NoOvnPhrase - } - - // Return response to client - response = &restapi.GetConstraintReferenceResponse{ - ConstraintReference: *constraint.ToRest(), - } - - return nil - } - - _, err = a.Store.Transact(ctx, dssstore.NewFuncOperation(action)) + response, err := dssstore.TransactWithResult[repos.Repository, *restapi.GetConstraintReferenceResponse](ctx, a.Store, req) if err != nil { err = stacktrace.Propagate(err, "Could not get constraint") if stacktrace.GetCode(err) == dsserr.NotFound { @@ -404,37 +334,13 @@ func (a *Server) QueryConstraintReferences(ctx context.Context, req *restapi.Que } // Parse area of interest to common Volume4D - vol4, err := scdmodels.Volume4DFromSCDRest(aoi) + _, err := scdmodels.Volume4DFromSCDRest(aoi) if err != nil { return restapi.QueryConstraintReferencesResponseSet{Response400: &restapi.ErrorResponse{ Message: dsserr.Handle(ctx, stacktrace.PropagateWithCode(err, dsserr.BadRequest, "Failed to convert to internal geometry model"))}} } - var response *restapi.QueryConstraintReferencesResponse - action := func(ctx context.Context, r repos.Repository) (err error) { - // Perform search query on Store - constraints, err := r.SearchConstraints(ctx, vol4) - if err != nil { - return err - } - - // Create response for client - response = &restapi.QueryConstraintReferencesResponse{ - ConstraintReferences: make([]restapi.ConstraintReference, 0, len(constraints)), - } - for _, constraint := range constraints { - p := constraint.ToRest() - if constraint.Manager != dssmodels.Manager(*req.Auth.ClientID) { - noOvnPhrase := restapi.EntityOVN(scdmodels.NoOvnPhrase) - p.Ovn = &noOvnPhrase - } - response.ConstraintReferences = append(response.ConstraintReferences, *p) - } - - return nil - } - - _, err = a.Store.Transact(ctx, dssstore.NewFuncOperation(action)) + response, err := dssstore.TransactWithResult[repos.Repository, *restapi.QueryConstraintReferencesResponse](ctx, a.Store, req) if err != nil { return restapi.QueryConstraintReferencesResponseSet{Response500: &api.InternalServerErrorBody{ ErrorMessage: *dsserr.Handle(ctx, stacktrace.Propagate(err, "Got an unexpected error"))}} 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..54c76e6c7 100644 --- a/pkg/scd/subscriptions_handler.go +++ b/pkg/scd/subscriptions_handler.go @@ -2,9 +2,7 @@ package scd import ( "context" - "time" - "github.com/golang/geo/s2" "github.com/interuss/dss/pkg/api" restapi "github.com/interuss/dss/pkg/api/scdv1" dsserr "github.com/interuss/dss/pkg/errors" @@ -14,11 +12,6 @@ import ( "github.com/interuss/dss/pkg/scd/repos" dssstore "github.com/interuss/dss/pkg/store" "github.com/interuss/stacktrace" - "github.com/jonboulle/clockwork" -) - -var ( - DefaultClock = clockwork.NewRealClock() ) func (a *Server) CreateSubscription(ctx context.Context, req *restapi.CreateSubscriptionRequest, @@ -32,8 +25,11 @@ func (a *Server) CreateSubscription(ctx context.Context, req *restapi.CreateSubs return restapi.CreateSubscriptionResponseSet{Response403: &restapi.ErrorResponse{ Message: dsserr.Handle(ctx, stacktrace.NewErrorWithCode(dsserr.PermissionDenied, "Missing owner"))}} } + if err := a.validatePutSubscriptionParams(req.Subscriptionid, req.Body); err != nil { + return restapi.CreateSubscriptionResponseSet{Response400: &restapi.ErrorResponse{Message: dsserr.Handle(ctx, err)}} + } - res, err := a.PutSubscription(ctx, *req.Auth.ClientID, req.Subscriptionid, "", req.Body) + res, err := dssstore.TransactWithResult[repos.Repository, *restapi.PutSubscriptionResponse](ctx, a.Store, req) if err != nil { err = stacktrace.Propagate(err, "Could not put subscription") errResp := &restapi.ErrorResponse{Message: dsserr.Handle(ctx, err)} @@ -64,8 +60,11 @@ func (a *Server) UpdateSubscription(ctx context.Context, req *restapi.UpdateSubs return restapi.UpdateSubscriptionResponseSet{Response403: &restapi.ErrorResponse{ Message: dsserr.Handle(ctx, stacktrace.NewErrorWithCode(dsserr.PermissionDenied, "Missing owner"))}} } + if err := a.validatePutSubscriptionParams(req.Subscriptionid, req.Body); err != nil { + return restapi.UpdateSubscriptionResponseSet{Response400: &restapi.ErrorResponse{Message: dsserr.Handle(ctx, err)}} + } - res, err := a.PutSubscription(ctx, *req.Auth.ClientID, req.Subscriptionid, req.Version, req.Body) + res, err := dssstore.TransactWithResult[repos.Repository, *restapi.PutSubscriptionResponse](ctx, a.Store, req) if err != nil { err = stacktrace.Propagate(err, "Could not put subscription") errResp := &restapi.ErrorResponse{Message: dsserr.Handle(ctx, err)} @@ -85,20 +84,17 @@ func (a *Server) UpdateSubscription(ctx context.Context, req *restapi.UpdateSubs return restapi.UpdateSubscriptionResponseSet{Response200: res} } -// PutSubscription creates a single subscription. -func (a *Server) PutSubscription(ctx context.Context, manager string, subscriptionid restapi.SubscriptionID, version string, params *restapi.PutSubscriptionParameters, -) (*restapi.PutSubscriptionResponse, error) { +// validatePutSubscriptionParams performs the request validation that can be done ahead of the transaction +func (a *Server) validatePutSubscriptionParams(subscriptionid restapi.SubscriptionID, params *restapi.PutSubscriptionParameters) error { // Retrieve Subscription ID - id, err := dssmodels.IDFromString(string(subscriptionid)) - - if err != nil { - return nil, stacktrace.NewErrorWithCode(dsserr.BadRequest, "Invalid ID format: `%s`", subscriptionid) + if _, err := dssmodels.IDFromString(string(subscriptionid)); err != nil { + return stacktrace.NewErrorWithCode(dsserr.BadRequest, "Invalid ID format: `%s`", subscriptionid) } if !a.AllowHTTPBaseUrls { - err = scdmodels.ValidateUSSBaseURL(string(params.UssBaseUrl)) + err := scdmodels.ValidateUSSBaseURL(string(params.UssBaseUrl)) if err != nil { - return nil, stacktrace.PropagateWithCode(err, dsserr.BadRequest, "Failed to validate base URL") + return stacktrace.PropagateWithCode(err, dsserr.BadRequest, "Failed to validate base URL") } } @@ -107,182 +103,29 @@ func (a *Server) PutSubscription(ctx context.Context, manager string, subscripti // If start time is not specified, it will default to the time the request is processed. extents, err := scdmodels.Volume4DFromSCDRest(¶ms.Extents) if err != nil { - return nil, stacktrace.PropagateWithCode(err, dsserr.BadRequest, "Unable to parse extents") + return stacktrace.PropagateWithCode(err, dsserr.BadRequest, "Unable to parse extents") } // Construct requested Subscription model - cells, err := extents.CalculateSpatialCovering() + _, err = extents.CalculateSpatialCovering() switch err { case nil, geo.ErrMissingSpatialVolume, geo.ErrMissingFootprint: // We may be able to fill these values from a previous Subscription or via defaults. default: - return nil, stacktrace.PropagateWithCode(err, dsserr.BadRequest, "Invalid area") + return stacktrace.PropagateWithCode(err, dsserr.BadRequest, "Invalid area") } - subreq := &scdmodels.Subscription{ - ID: id, - Manager: dssmodels.Manager(manager), - Version: scdmodels.OVN(version), - - StartTime: extents.StartTime, - EndTime: extents.EndTime, - AltitudeLo: extents.SpatialVolume.AltitudeLo, - AltitudeHi: extents.SpatialVolume.AltitudeHi, - Cells: cells, - - USSBaseURL: string(params.UssBaseUrl), - } - if params.NotifyForOperationalIntents != nil { - subreq.NotifyForOperationalIntents = *params.NotifyForOperationalIntents - } - if params.NotifyForConstraints != nil { - subreq.NotifyForConstraints = *params.NotifyForConstraints - } + notifyForOperationalIntents := params.NotifyForOperationalIntents != nil && *params.NotifyForOperationalIntents + notifyForConstraints := params.NotifyForConstraints != nil && *params.NotifyForConstraints // Validate requested Subscription - if !subreq.NotifyForOperationalIntents && !subreq.NotifyForConstraints { - return nil, stacktrace.NewErrorWithCode(dsserr.BadRequest, "No notification triggers requested for Subscription") + if !notifyForOperationalIntents && !notifyForConstraints { + return stacktrace.NewErrorWithCode(dsserr.BadRequest, "No notification triggers requested for Subscription") } // TODO: Check scopes to verify requested information (op intents or constraints) may be requested - var result *restapi.PutSubscriptionResponse - action := func(ctx context.Context, r repos.Repository) (err error) { - // Check existing Subscription (if any) - old, err := r.GetSubscription(ctx, subreq.ID) - if err != nil { - return stacktrace.Propagate(err, "Could not get Subscription from repo") - } - - // Validate and perhaps correct StartTime and EndTime. - if err := subreq.AdjustTimeRange(DefaultClock.Now(), old); err != nil { - return stacktrace.Propagate(err, "Error adjusting time range of Subscription") - } - - var dependentOpIds []dssmodels.ID - - if old == nil { - // There is no previous Subscription (this is a creation attempt) - if subreq.Version.String() != "" { - // The user wants to update an existing Subscription, but one wasn't found. - return stacktrace.NewErrorWithCode(dsserr.NotFound, "Subscription %s not found", subreq.ID.String()) - } - } else { - // There is a previous Subscription (this is an update attempt) - switch { - case subreq.Version.String() == "": - // The user wants to create a new Subscription but it already exists. - return stacktrace.NewErrorWithCode(dsserr.AlreadyExists, "Subscription %s already exists", subreq.ID.String()) - case subreq.Version.String() != old.Version.String(): - // The user wants to update a Subscription but the version doesn't match. - return stacktrace.Propagate( - stacktrace.NewErrorWithCode(dsserr.VersionMismatch, "Subscription version %s is not current", subreq.Version), - "Current version is %s but client specified version %s", old.Version, subreq.Version) - case old.Manager != subreq.Manager: - return stacktrace.Propagate( - stacktrace.NewErrorWithCode(dsserr.PermissionDenied, "Subscription is owned by different client"), - "Subscription owned by %s, but %s attempted to modify", old.Manager, subreq.Manager) - } - - subreq.NotificationIndex = old.NotificationIndex - - // Validate Subscription against DependentOperations - dependentOpIds, err = r.GetDependentOperationalIntents(ctx, subreq.ID) - if err != nil { - return stacktrace.Propagate(err, "Could not find dependent Operation Ids") - } - - operations, err := GetOperations(ctx, r, dependentOpIds) - if err != nil { - return stacktrace.Propagate(err, "Could not get all dependent Operations") - } - if err := subreq.ValidateDependentOps(operations); err != nil { - // The provided subscription does not cover all its dependent operations - return err - } - } - - // Store Subscription model - sub, err := r.UpsertSubscription(ctx, subreq) - if err != nil { - return stacktrace.Propagate(err, "Could not upsert Subscription into repo") - } - if sub == nil { - return stacktrace.NewError("UpsertSubscription returned no Subscription for ID: %s", id) - } - - // Convert Subscription to REST - p, err := sub.ToRest(dependentOpIds) - if err != nil { - return stacktrace.Propagate(err, "Could not convert Subscription to REST model") - } - result = &restapi.PutSubscriptionResponse{ - Subscription: *p, - } - - if sub.NotifyForOperationalIntents { - // Find relevant Operations - var relevantOperations []*scdmodels.OperationalIntent - if len(sub.Cells) > 0 { - ops, err := r.SearchOperationalIntents(ctx, &dssmodels.Volume4D{ - StartTime: sub.StartTime, - EndTime: sub.EndTime, - SpatialVolume: &dssmodels.Volume3D{ - AltitudeLo: sub.AltitudeLo, - AltitudeHi: sub.AltitudeHi, - Footprint: dssmodels.GeometryFunc(func() (s2.CellUnion, error) { - return sub.Cells, nil - }), - }, - }) - if err != nil { - return stacktrace.Propagate(err, "Could not search Operations in repo") - } - relevantOperations = ops - } - // Attach Operations to response - opIntentRefs := make([]restapi.OperationalIntentReference, 0, len(relevantOperations)) - for _, op := range relevantOperations { - if op.Manager != dssmodels.Manager(manager) { - op.OVN = scdmodels.NoOvnPhrase - } - - opIntentRefs = append(opIntentRefs, *op.ToRest()) - } - result.OperationalIntentReferences = &opIntentRefs - } - - if sub.NotifyForConstraints { - // Query relevant Constraints - constraints, err := r.SearchConstraints(ctx, extents) - if err != nil { - return stacktrace.Propagate(err, "Could not search Constraints in repo") - } - - // Attach Constraints to response - constraintRefs := make([]restapi.ConstraintReference, 0, len(constraints)) - for _, constraint := range constraints { - p := constraint.ToRest() - if constraint.Manager != dssmodels.Manager(manager) { - noOvnPhrase := restapi.EntityOVN(scdmodels.NoOvnPhrase) - p.Ovn = &noOvnPhrase - } - - constraintRefs = append(constraintRefs, *p) - } - result.ConstraintReferences = &constraintRefs - } - - return nil - } - - _, err = a.Store.Transact(ctx, dssstore.NewFuncOperation(action)) - if err != nil { - return nil, err // No need to Propagate this error as this is not a useful stacktrace line - } - - // Return response to client - return result, nil + return nil } // GetSubscription returns a single subscription for the given ID. @@ -290,7 +133,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 +145,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 +168,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 +182,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 +194,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 +220,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 +239,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)} @@ -538,16 +260,3 @@ func (a *Server) DeleteSubscription(ctx context.Context, req *restapi.DeleteSubs return restapi.DeleteSubscriptionResponseSet{Response200: response} } - -// GetOperations gets operations by given ids -func GetOperations(ctx context.Context, r repos.Repository, opIDs []dssmodels.ID) ([]*scdmodels.OperationalIntent, error) { - var res []*scdmodels.OperationalIntent - for _, opID := range opIDs { - operation, err := r.GetOperationalIntent(ctx, opID) - if err != nil { - return nil, stacktrace.Propagate(err, "Could not retrieve dependent Operation %s", opID) - } - res = append(res, operation) - } - return res, nil -} diff --git a/pkg/scd/uss_availability_handler.go b/pkg/scd/uss_availability_handler.go index 607876afc..d8f7ced19 100644 --- a/pkg/scd/uss_availability_handler.go +++ b/pkg/scd/uss_availability_handler.go @@ -8,22 +8,13 @@ import ( restapi "github.com/interuss/dss/pkg/api/scdv1" dsserr "github.com/interuss/dss/pkg/errors" dssmodels "github.com/interuss/dss/pkg/models" + "github.com/interuss/dss/pkg/scd/actions" 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/stacktrace" - "github.com/jackc/pgx/v5" ) -func GetDefaultAvailabilityResponse(id dssmodels.Manager) *restapi.UssAvailabilityStatusResponse { - return &restapi.UssAvailabilityStatusResponse{ - Status: restapi.UssAvailabilityStatus{ - Availability: restapi.UssAvailabilityState_Unknown, - Uss: id.String()}, - Version: "", - } -} - func (a *Server) GetUssAvailability(ctx context.Context, req *restapi.GetUssAvailabilityRequest, ) restapi.GetUssAvailabilityResponseSet { @@ -33,30 +24,11 @@ func (a *Server) GetUssAvailability(ctx context.Context, req *restapi.GetUssAvai Message: dsserr.Handle(ctx, stacktrace.NewErrorWithCode(dsserr.BadRequest, "UssId not provided"))}} } - var response *restapi.UssAvailabilityStatusResponse - action := func(ctx context.Context, r repos.Repository) (err error) { - // Get USS availability from Store - ussa, err := r.GetUssAvailability(ctx, id) - if err != nil && err != pgx.ErrNoRows { - return stacktrace.Propagate(err, "Could not get USS availability from repo") - } - if ussa == nil { - // Return default availability status "Unknown" - response = GetDefaultAvailabilityResponse(id) - return nil - } - response = &restapi.UssAvailabilityStatusResponse{ - Status: *ussa.ToRest(), - Version: ussa.Version.String(), - } - return nil - } - - _, err := a.Store.Transact(ctx, dssstore.NewFuncOperation(action)) + response, err := dssstore.TransactWithResult[repos.Repository, *restapi.UssAvailabilityStatusResponse](ctx, a.Store, req) if err != nil { // In case of older DB versions where availability table doesn't exist if strings.Contains(err.Error(), "does not exist") { - response = GetDefaultAvailabilityResponse(id) + response = actions.GetDefaultAvailabilityResponse(id) } else { // No need to Propagate this error as this is not a useful stacktrace line return restapi.GetUssAvailabilityResponseSet{Response500: &api.InternalServerErrorBody{ @@ -79,54 +51,17 @@ func (a *Server) SetUssAvailability(ctx context.Context, req *restapi.SetUssAvai } // Retrieve USS availability status from request params - availability, err := scdmodels.UssAvailabilityStateFromRest(req.Body.Availability) + _, err := scdmodels.UssAvailabilityStateFromRest(req.Body.Availability) if err != nil { return restapi.SetUssAvailabilityResponseSet{Response400: &restapi.ErrorResponse{ Message: dsserr.Handle(ctx, stacktrace.PropagateWithCode(err, dsserr.BadRequest, "Invalid availability state"))}} } - id := dssmodels.ManagerFromString(req.UssId) - version := scdmodels.OVN(req.Body.OldVersion) - ussareq := &scdmodels.UssAvailabilityStatus{ - Uss: id, - Availability: availability, - } - - var result *restapi.UssAvailabilityStatusResponse - action := func(ctx context.Context, r repos.Repository) (err error) { - old, err := r.GetUssAvailability(ctx, id) - if err != nil && err != pgx.ErrNoRows { - return stacktrace.Propagate(err, "Could not get USS availability from repo") - } - switch { - case old == nil && !version.Empty(): - // The user wants set a new availability status but it already exists. - return stacktrace.NewErrorWithCode(dsserr.AlreadyExists, "availability for USS %s already exists", id.String()) - case old != nil && old.Version != version: - // The user wants to update an availability status but the version doesn't match. - return stacktrace.Propagate( - stacktrace.NewErrorWithCode(dsserr.VersionMismatch, "USS availability version %s is not current", version), - "Current version is %s but client specified version %s", old.Version, version) - } - // Upsert the USS availability - ussa, err := r.UpsertUssAvailability(ctx, ussareq) - if err != nil { - return stacktrace.Propagate(err, "Could not upsert USS Availability into repo") - } - if ussa == nil { - return stacktrace.NewError("UpsertUssAvailability returned no USS availability for ID: %s", id) - } - result = &restapi.UssAvailabilityStatusResponse{ - Status: *ussa.ToRest(), - Version: ussa.Version.String(), - } - return nil - } - _, err = a.Store.Transact(ctx, dssstore.NewFuncOperation(action)) + result, err := dssstore.TransactWithResult[repos.Repository, *restapi.UssAvailabilityStatusResponse](ctx, a.Store, req) if err != nil { // In case of older DB versions where availability table doesn't exist if strings.Contains(err.Error(), "does not exist") { - result = GetDefaultAvailabilityResponse(id) + result = actions.GetDefaultAvailabilityResponse(dssmodels.ManagerFromString(req.UssId)) } else { err = stacktrace.Propagate(err, "Could not set USS availability status") errResp := &restapi.ErrorResponse{Message: dsserr.Handle(ctx, err)}