[raft/scd] Add scd Raftstore subscriptions

Author: MariemBaccari

pkg/scd/store/raftstore/store.go

@@ -32,6 +32,20 @@ const (
 	GetOperationalIntentTransaction    raftstore.RequestType = "getOperationalIntentTransaction"
 	QueryOperationalIntentTransaction  raftstore.RequestType = "queryOperationalIntentTransaction"
 	UpsertOperationalIntentTransaction raftstore.RequestType = "upsertOperationalIntentTransaction"
+
+	searchSubscriptions                 raftstore.RequestType = "searchSubscriptions"
+	getSubscription                     raftstore.RequestType = "getSubscription"
+	upsertSubscription                  raftstore.RequestType = "upsertSubscription"
+	deleteSubscription                  raftstore.RequestType = "deleteSubscription"
+	incrementNotificationForOIs         raftstore.RequestType = "incrementNotificationForOIs"
+	incrementNotificationForConstraints raftstore.RequestType = "incrementNotificationForConstraints"
+	listExpiredSubscriptions            raftstore.RequestType = "listExpiredSubscriptions"
+	countSubscriptions                  raftstore.RequestType = "countSubscriptions"
+
+	DeleteSubscriptionTransaction raftstore.RequestType = "deleteSubscriptionTransaction"
+	GetSubscriptionTransaction    raftstore.RequestType = "getSubscriptionTransaction"
+	QuerySubscriptionTransaction  raftstore.RequestType = "querySubscriptionTransaction"
+	UpsertSubscriptionTransaction raftstore.RequestType = "upsertSubscriptionTransaction"
 )
 
 var readOnlyRequests = []raftstore.RequestType{
@@ -43,6 +57,14 @@ var readOnlyRequests = []raftstore.RequestType{
 
 	GetOperationalIntentTransaction,
 	QueryOperationalIntentTransaction,
+
+	searchSubscriptions,
+	getSubscription,
+	listExpiredSubscriptions,
+	countSubscriptions,
+
+	GetSubscriptionTransaction,
+	QuerySubscriptionTransaction,
 }
 
 // repo is a full implementation of scd.repos.Repository for Raft-based storage.
@@ -158,6 +180,77 @@ func (r *repo) Apply(ctx context.Context, proposal consensus.Proposal) (any, err
 	case UpsertOperationalIntentTransaction:
 		return r.upsertOperationalIntentTransactionApplier(ctx, proposal, mem)
 
+	case searchSubscriptions:
+		var v4d *dssmodels.Volume4D
+		if err := json.Unmarshal(proposal.Value, &v4d); err != nil {
+			return nil, stacktrace.Propagate(err, "failed to unmarshal %s proposal value", searchSubscriptions)
+		}
+
+		return mem.SearchSubscriptions(ctx, v4d)
+
+	case getSubscription:
+		var id dssmodels.ID
+		if err := json.Unmarshal(proposal.Value, &id); err != nil {
+			return nil, stacktrace.Propagate(err, "failed to unmarshal %s proposal value", getSubscription)
+		}
+
+		return mem.GetSubscription(ctx, id)
+
+	case upsertSubscription:
+		var sub *scdmodels.Subscription
+		if err := json.Unmarshal(proposal.Value, &sub); err != nil {
+			return nil, stacktrace.Propagate(err, "failed to unmarshal %s proposal value", upsertSubscription)
+		}
+
+		return mem.UpsertSubscription(ctx, sub)
+
+	case deleteSubscription:
+		var id dssmodels.ID
+		if err := json.Unmarshal(proposal.Value, &id); err != nil {
+			return nil, stacktrace.Propagate(err, "failed to unmarshal %s proposal value", deleteSubscription)
+		}
+
+		return nil, mem.DeleteSubscription(ctx, id)
+
+	case incrementNotificationForOIs:
+		var v4d *dssmodels.Volume4D
+		if err := json.Unmarshal(proposal.Value, &v4d); err != nil {
+			return nil, stacktrace.Propagate(err, "failed to unmarshal %s proposal value", incrementNotificationForOIs)
+		}
+
+		return mem.IncrementNotificationIndicesForOperationalIntents(ctx, v4d)
+
+	case incrementNotificationForConstraints:
+		var v4d *dssmodels.Volume4D
+		if err := json.Unmarshal(proposal.Value, &v4d); err != nil {
+			return nil, stacktrace.Propagate(err, "failed to unmarshal %s proposal value", incrementNotificationForConstraints)
+		}
+
+		return mem.IncrementNotificationIndicesForConstraints(ctx, v4d)
+
+	case listExpiredSubscriptions:
+		var threshold time.Time
+		if err := json.Unmarshal(proposal.Value, &threshold); err != nil {
+			return nil, stacktrace.Propagate(err, "failed to unmarshal %s proposal value", listExpiredSubscriptions)
+		}
+
+		return mem.ListExpiredSubscriptions(ctx, threshold)
+
+	case countSubscriptions:
+		return mem.CountSubscriptions(ctx)
+
+	case UpsertSubscriptionTransaction:
+		return r.upsertSubscriptionTransactionApplier(ctx, proposal, mem)
+
+	case DeleteSubscriptionTransaction:
+		return r.deleteSubscriptionTransactionApplier(ctx, proposal, mem)
+
+	case GetSubscriptionTransaction:
+		return r.getSubscriptionTransactionApplier(ctx, proposal, mem)
+
+	case QuerySubscriptionTransaction:
+		return r.querySubscriptionTransactionApplier(ctx, proposal, mem)
+
 	default:
 		return nil, stacktrace.NewError("unknown request type: %q", proposal.RequestType)
 	}

pkg/scd/store/raftstore/subscriptions.go

@@ -5,44 +5,143 @@ import (
 	"time"
 
 	"github.com/golang/geo/s2"
-	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/stacktrace"
 )
 
-func (r *repo) SearchSubscriptions(_ context.Context, v4d *dssmodels.Volume4D) ([]*scdmodels.Subscription, error) {
-	return nil, stacktrace.NewErrorWithCode(dsserr.NotImplemented, "SearchSubscriptions not implemented for raftstore")
+func (r *repo) SearchSubscriptions(ctx context.Context, v4d *dssmodels.Volume4D) ([]*scdmodels.Subscription, error) {
+	result, err := r.consensus.ProposeValue(ctx, storeID, string(searchSubscriptions), v4d, true)
+	if err != nil {
+		return nil, stacktrace.Propagate(err, "failed to propose searchSubscriptions")
+	}
+
+	if result == nil {
+		return nil, nil
+	}
+
+	subs, ok := result.([]*scdmodels.Subscription)
+	if !ok {
+		return nil, stacktrace.NewError("invalid result type: %T", result)
+	}
+
+	return subs, nil
 }
 
-func (r *repo) GetSubscription(_ context.Context, id dssmodels.ID) (*scdmodels.Subscription, error) {
-	return nil, stacktrace.NewErrorWithCode(dsserr.NotImplemented, "GetSubscription not implemented for raftstore")
+func (r *repo) GetSubscription(ctx context.Context, id dssmodels.ID) (*scdmodels.Subscription, error) {
+	result, err := r.consensus.ProposeValue(ctx, storeID, string(getSubscription), id, true)
+	if err != nil {
+		return nil, stacktrace.Propagate(err, "failed to propose getSubscription")
+	}
+
+	if result == nil {
+		return nil, nil
+	}
+
+	sub, ok := result.(*scdmodels.Subscription)
+	if !ok {
+		return nil, stacktrace.NewError("invalid result type: %T", result)
+	}
+
+	return sub, nil
 }
 
-func (r *repo) UpsertSubscription(_ context.Context, sub *scdmodels.Subscription) (*scdmodels.Subscription, error) {
-	return nil, stacktrace.NewErrorWithCode(dsserr.NotImplemented, "UpsertSubscription not implemented for raftstore")
+func (r *repo) UpsertSubscription(ctx context.Context, sub *scdmodels.Subscription) (*scdmodels.Subscription, error) {
+	result, err := r.consensus.ProposeValue(ctx, storeID, string(upsertSubscription), sub, false)
+	if err != nil {
+		return nil, stacktrace.Propagate(err, "failed to propose upsertSubscription")
+	}
+
+	if result == nil {
+		return nil, nil
+	}
+
+	upserted, ok := result.(*scdmodels.Subscription)
+	if !ok {
+		return nil, stacktrace.NewError("invalid result type: %T", result)
+	}
+
+	return upserted, nil
 }
 
-func (r *repo) DeleteSubscription(_ context.Context, id dssmodels.ID) error {
-	return stacktrace.NewErrorWithCode(dsserr.NotImplemented, "DeleteSubscription not implemented for raftstore")
+func (r *repo) DeleteSubscription(ctx context.Context, id dssmodels.ID) error {
+	_, err := r.consensus.ProposeValue(ctx, storeID, string(deleteSubscription), id, false)
+	return err
 }
 
-func (r *repo) IncrementNotificationIndicesForOperationalIntents(_ context.Context, v4d *dssmodels.Volume4D) ([]*scdmodels.Subscription, error) {
-	return nil, stacktrace.NewErrorWithCode(dsserr.NotImplemented, "IncrementNotificationIndicesForOperationalIntents not implemented for raftstore")
+func (r *repo) IncrementNotificationIndicesForOperationalIntents(ctx context.Context, v4d *dssmodels.Volume4D) ([]*scdmodels.Subscription, error) {
+	result, err := r.consensus.ProposeValue(ctx, storeID, string(incrementNotificationForOIs), v4d, false)
+	if err != nil {
+		return nil, stacktrace.Propagate(err, "failed to propose incrementNotificationForOIs")
+	}
+
+	if result == nil {
+		return nil, nil
+	}
+
+	subs, ok := result.([]*scdmodels.Subscription)
+	if !ok {
+		return nil, stacktrace.NewError("invalid result type: %T", result)
+	}
+
+	return subs, nil
 }
 
-func (r *repo) IncrementNotificationIndicesForConstraints(_ context.Context, v4d *dssmodels.Volume4D) ([]*scdmodels.Subscription, error) {
-	return nil, stacktrace.NewErrorWithCode(dsserr.NotImplemented, "IncrementNotificationIndicesForConstraints not implemented for raftstore")
+func (r *repo) IncrementNotificationIndicesForConstraints(ctx context.Context, v4d *dssmodels.Volume4D) ([]*scdmodels.Subscription, error) {
+	result, err := r.consensus.ProposeValue(ctx, storeID, string(incrementNotificationForConstraints), v4d, false)
+	if err != nil {
+		return nil, stacktrace.Propagate(err, "failed to propose incrementNotificationForConstraints")
+	}
+
+	if result == nil {
+		return nil, nil
+	}
+
+	subs, ok := result.([]*scdmodels.Subscription)
+	if !ok {
+		return nil, stacktrace.NewError("invalid result type: %T", result)
+	}
+
+	return subs, nil
 }
 
-func (r *repo) LockSubscriptionsOnCells(_ context.Context, cells s2.CellUnion, subscriptionIds []dssmodels.ID, startTime *time.Time, endTime *time.Time) error {
-	return stacktrace.NewErrorWithCode(dsserr.NotImplemented, "LockSubscriptionsOnCells not implemented for raftstore")
+func (r *repo) LockSubscriptionsOnCells(_ context.Context, _ s2.CellUnion, _ []dssmodels.ID, _ *time.Time, _ *time.Time) error {
+	// for the raftstore, LockSubscriptionsOnCells is a no-op
+	return nil
 }
 
-func (r *repo) ListExpiredSubscriptions(_ context.Context, threshold time.Time) ([]*scdmodels.Subscription, error) {
-	return nil, stacktrace.NewErrorWithCode(dsserr.NotImplemented, "ListExpiredSubscriptions not implemented for raftstore")
+func (r *repo) ListExpiredSubscriptions(ctx context.Context, threshold time.Time) ([]*scdmodels.Subscription, error) {
+	result, err := r.consensus.ProposeValue(ctx, storeID, string(listExpiredSubscriptions), threshold, true)
+	if err != nil {
+		return nil, stacktrace.Propagate(err, "failed to propose listExpiredSubscriptions")
+	}
+
+	if result == nil {
+		return nil, nil
+	}
+
+	subs, ok := result.([]*scdmodels.Subscription)
+	if !ok {
+		return nil, stacktrace.NewError("invalid result type: %T", result)
+	}
+
+	return subs, nil
 }
 
-func (r *repo) CountSubscriptions(_ context.Context) (int64, error) {
-	return 0, stacktrace.NewErrorWithCode(dsserr.NotImplemented, "CountSubscriptions not implemented for raftstore")
+func (r *repo) CountSubscriptions(ctx context.Context) (int64, error) {
+	result, err := r.consensus.ProposeValue(ctx, storeID, string(countSubscriptions), nil, true)
+	if err != nil {
+		return 0, stacktrace.Propagate(err, "failed to propose countSubscriptions")
+	}
+
+	if result == nil {
+		return 0, nil
+	}
+
+	count, ok := result.(int64)
+	if !ok {
+		return 0, stacktrace.NewError("invalid result type: %T", result)
+	}
+
+	return count, nil
 }