Skip to content

Commit

Permalink
[scd] properly cleanup implicit sub on oir update
Browse files Browse the repository at this point in the history 
This is a rework of interuss#1057 after most changes for interuss#1088 have landed.

The DSS built from this PR successfully passed the new qualifier's implicit subscription handling scenario, although that scenario still needs to be extended to cover all corner cases.

The PR is open for review, as it addresses interuss#1056
  • Loading branch information
@Shastick
Shastick committed Sep 9, 2024
1 parent 60b2b3c commit 0c7b2c3
Showing 1 changed file with 94 additions and 34 deletions.
128 changes: 94 additions & 34 deletions pkg/scd/operational_intents_handler.go
View file
Original file line number Diff line number Diff line change
Expand Up @@ -16,6 +16,37 @@ import (
"github.com/interuss/stacktrace"
)

// subscriptionIsImplicitAndOnlyAttachedToOIR will check if:
// - the subscription is defined and is implicit
// - the subscription is attached to the specified operational intent
// - the subscription is not attached to any other operational intent
//
// This is to be used in contexts where an implicit subscription may need to be cleaned up: if true is returned,
// the subscription can be safely removed after the operational intent is deleted or attached to another subscription.
//
// NOTE: this should eventually be pushed down to CRDB as part of the queries being executed in the callers of this method.
//
// See https://github.com/interuss/dss/issues/1059 for more details
func subscriptionIsImplicitAndOnlyAttachedToOIR(ctx context.Context, r repos.Repository, oirID dssmodels.ID, subscription *scdmodels.Subscription) (bool, error) {
if subscription == nil {
return false, nil
}
if !subscription.ImplicitSubscription {
return false, nil
}
// Get the Subscription's dependent OperationalIntents
dependentOps, err := r.GetDependentOperationalIntents(ctx, subscription.ID)
if err != nil {
return false, stacktrace.Propagate(err, "Could not find dependent OperationalIntents")
}
if len(dependentOps) == 0 {
return false, stacktrace.NewError("An implicit Subscription had no dependent OperationalIntents")
} else if len(dependentOps) == 1 && dependentOps[0] == oirID {
return true, nil
}
return false, nil
}

// DeleteOperationalIntentReference deletes a single operational intent ref for a given ID at
// the specified version.
func (a *Server) DeleteOperationalIntentReference(ctx context.Context, req *restapi.DeleteOperationalIntentReferenceRequest,
Expand Down Expand Up @@ -69,29 +100,20 @@ func (a *Server) DeleteOperationalIntentReference(ctx context.Context, req *rest
}

// Get the Subscription supporting the OperationalIntent, if one is defined
var sub *scdmodels.Subscription
removeImplicitSubscription := false
var previousSubscription *scdmodels.Subscription
if old.SubscriptionID != nil {
sub, err = r.GetSubscription(ctx, *old.SubscriptionID)
previousSubscription, err = r.GetSubscription(ctx, *old.SubscriptionID)
if err != nil {
return stacktrace.Propagate(err, "Unable to get OperationalIntent's Subscription from repo")
}
if sub == nil {
if previousSubscription == nil {
return stacktrace.NewError("OperationalIntent's Subscription missing from repo")
}
}

if sub.ImplicitSubscription {
// Get the Subscription's dependent OperationalIntents
dependentOps, err := r.GetDependentOperationalIntents(ctx, sub.ID)
if err != nil {
return stacktrace.Propagate(err, "Could not find dependent OperationalIntents")
}
if len(dependentOps) == 0 {
return stacktrace.NewError("An implicit Subscription had no dependent OperationalIntents")
} else if len(dependentOps) == 1 {
removeImplicitSubscription = true
}
}
removeImplicitSubscription, err := subscriptionIsImplicitAndOnlyAttachedToOIR(ctx, r, id, previousSubscription)
if err != nil {
return stacktrace.Propagate(err, "Could not determine if Subscription can be removed")
}

// Gather the subscriptions that need to be notified
Expand Down Expand Up @@ -119,7 +141,7 @@ func (a *Server) DeleteOperationalIntentReference(ctx context.Context, req *rest
// removeImplicitSubscription is only true if the OIR had a subscription defined
if removeImplicitSubscription {
// Automatically remove a now-unused implicit Subscription
err = r.DeleteSubscription(ctx, sub.ID)
err = r.DeleteSubscription(ctx, previousSubscription.ID)
if err != nil {
return stacktrace.Propagate(err, "Unable to delete associated implicit Subscription")
}
Expand Down Expand Up @@ -810,33 +832,63 @@ func (a *Server) upsertOperationalIntentReference(ctx context.Context, authorize

// For an OIR being created, version starts at 0
version := int32(0)
var previousSub *scdmodels.Subscription
if old != nil {
version = int32(old.Version)
// Fetch the previous OIR's subscription if it exists
if old.SubscriptionID != nil {
previousSub, err = r.GetSubscription(ctx, *old.SubscriptionID)
if err != nil {
return stacktrace.Propagate(err, "Unable to get OperationalIntent's Subscription from repo")
}
}
}

var sub *scdmodels.Subscription
// Determine if the previous subscription is being replaced and if it will need to be cleaned up
previousSubIsBeingReplaced := previousSub != nil && validParams.subscriptionID != previousSub.ID
removePreviousImplicitSubscription := false
if previousSubIsBeingReplaced {
removePreviousImplicitSubscription, err = subscriptionIsImplicitAndOnlyAttachedToOIR(ctx, r, validParams.id, previousSub)
if err != nil {
return stacktrace.Propagate(err, "Could not determine if previous Subscription can be removed")
}
}

// attachedSub is the subscription that will end up being attached to the OIR
// it defaults to the previous subscription (which may be nil), and may be updated if required by the parameters
attachedSub := previousSub
if validParams.subscriptionID.Empty() {
// Create an implicit subscription if one has been requested.
// Requesting neither an explicit nor an implicit subscription is allowed for ACCEPTED states:
// for other states, an error will have been returned earlier.
// if no implicit subscription is requested and we reached this point, we will proceed without subscription
// No subscription ID was provided:
// check if an implicit subscription should be created, otherwise do nothing
if validParams.implicitSubscription.requested {
if sub, err = createAndStoreNewImplicitSubscription(ctx, r, manager, validParams); err != nil {
// Parameters for a new implicit subscription have been passed: we will create
// a new implicit subscription even if another subscription was attached to this OIR before,
// regardless of whether it was an implicit subscription or not.
if attachedSub, err = createAndStoreNewImplicitSubscription(ctx, r, manager, validParams); err != nil {
return stacktrace.Propagate(err, "Failed to create implicit subscription")
}
} else {
// If no subscription ID is provided and no implicit subscription is requested,
// the OIR should have no attached subscription
attachedSub = nil
}
} else {
// Attempt to rely on the specified subscription
sub, err = r.GetSubscription(ctx, validParams.subscriptionID)
if err != nil {
return stacktrace.Propagate(err, "Unable to get depended-upon Subscription")
}
if sub == nil {
return stacktrace.NewErrorWithCode(dsserr.BadRequest, "Depended-upon Subscription %s does not exist", validParams.subscriptionID)
// If it is different from the previous subscription, we need to fetch it from the store
// in order to ensure it correctly covers the OIR.
// We do the check below in order to avoid re-fetching the subscription if it has not changed
if attachedSub == nil || previousSubIsBeingReplaced {
attachedSub, err = r.GetSubscription(ctx, validParams.subscriptionID)
if err != nil {
return stacktrace.Propagate(err, "Unable to get requested Subscription from store")
}
if attachedSub == nil {
return stacktrace.NewErrorWithCode(dsserr.BadRequest, "Specified Subscription %s does not exist", validParams.subscriptionID)
}
}

// We need to confirm that it is owned by the calling manager
if sub.Manager != manager {
if attachedSub.Manager != manager {
return stacktrace.Propagate(
// We do a bit of wrapping gymnastics because the root error message will be sent in the response,
// and we don't want to include the effective manager in there.
Expand All @@ -845,34 +897,42 @@ func (a *Server) upsertOperationalIntentReference(ctx context.Context, authorize
// The propagation message will end in the logs and help with debugging.
"Subscription %s owned by %s, but %s attempted to use it for an OperationalIntent",
validParams.subscriptionID,
sub.Manager,
attachedSub.Manager,
manager,
)
}

// We need to ensure the subscription covers the OIR's geo-temporal extent
sub, err = ensureSubscriptionCoversOIR(ctx, r, sub, validParams)
attachedSub, err = ensureSubscriptionCoversOIR(ctx, r, attachedSub, validParams)
if err != nil {
return stacktrace.Propagate(err, "Failed to ensure subscription covers OIR")
}
}

if validParams.state.RequiresKey() {
responseConflict, err = validateKeyAndProvideConflictResponse(ctx, r, manager, validParams, sub)
responseConflict, err = validateKeyAndProvideConflictResponse(ctx, r, manager, validParams, attachedSub)
if err != nil {
return stacktrace.PropagateWithCode(err, stacktrace.GetCode(err), "Failed to validate key")
}
}

// Construct the new OperationalIntent
op := validParams.toOIR(manager, sub, version+1)
op := validParams.toOIR(manager, attachedSub, version+1)

// Upsert the OperationalIntent
op, err = r.UpsertOperationalIntent(ctx, op)
if err != nil {
return stacktrace.Propagate(err, "Failed to upsert OperationalIntent in repo")
}

// Check if the previously attached subscription should be removed
if removePreviousImplicitSubscription {
err = r.DeleteSubscription(ctx, previousSub.ID)
if err != nil {
return stacktrace.Propagate(err, "Unable to delete previous implicit Subscription")
}
}

notifyVolume, err := computeNotificationVolume(old, validParams.uExtent)
if err != nil {
return stacktrace.Propagate(err, "Failed to compute notification volume")
Expand Down

0 comments on commit 0c7b2c3

Please sign in to comment.