allow only one ongoing SetIntentRPC per datastore

Author: karimra

pkg/datastore/datastore_rpc.go

@@ -5,6 +5,7 @@ import (
 	"errors"
 	"fmt"
 	"strings"
+	"sync"
 	"time"
 
 	"github.com/iptecharch/cache/proto/cachepb"
@@ -44,6 +45,12 @@ type Datastore struct {
 
 	// stop cancel func
 	cfn context.CancelFunc
+
+	// intent semaphore.
+	// Used by SetIntent to guarantee that
+	// only one SetIntent
+	// is applied at a time.
+	intentMutex *sync.Mutex
 }
 
 // New creates a new datastore, its schema server client and initializes the SBI target
@@ -53,6 +60,7 @@ func New(ctx context.Context, c *config.DatastoreConfig, scc schema.Client, cc c
 		config:       c,
 		schemaClient: scc,
 		cacheClient:  cc,
+		intentMutex:  new(sync.Mutex),
 	}
 	if c.Sync != nil {
 		ds.synCh = make(chan *target.SyncUpdate, c.Sync.Buffer)

pkg/datastore/intent_rpc.go

@@ -12,6 +12,8 @@ import (
 	"github.com/iptecharch/cache/proto/cachepb"
 	sdcpb "github.com/iptecharch/sdc-protos/sdcpb"
 	log "github.com/sirupsen/logrus"
+	"google.golang.org/grpc/codes"
+	"google.golang.org/grpc/status"
 	"google.golang.org/protobuf/encoding/prototext"
 	"google.golang.org/protobuf/proto"
 
@@ -45,6 +47,11 @@ func (d *Datastore) GetIntent(ctx context.Context, req *sdcpb.GetIntentRequest)
 }
 
 func (d *Datastore) SetIntent(ctx context.Context, req *sdcpb.SetIntentRequest) (*sdcpb.SetIntentResponse, error) {
+	if !d.intentMutex.TryLock() {
+		return nil, status.Errorf(codes.ResourceExhausted, "datastore %s has an ongoing SetIntentRequest", d.Name())
+	}
+	defer d.intentMutex.Unlock()
+
 	log.Infof("received SetIntentRequest: ds=%s intent=%s", req.GetName(), req.GetIntent())
 	now := time.Now().UnixNano()
 	candidateName := fmt.Sprintf("%s-%d", req.GetIntent(), now)