fix(datastore): repair transaction locking and auto-rollback

pkg/datastore/datastore_rpc.go

@@ -58,8 +58,10 @@ type Datastore struct {
 	m                *sync.RWMutex
 	deviationClients map[sdcpb.DataServer_WatchDeviationsServer]string
 
-	// dmutex serializes Datastore set-path work (TransactionSet/Confirm/Cancel
-	// and Sync Revert's southbound apply). Fail-fast TryLock; never block.
+	// dmutex serializes Datastore set-path work (TransactionSet/Cancel, the
+	// timer-driven transaction rollback and Sync Revert's southbound apply).
+	// Request handlers use a fail-fast TryLock and never block; only the
+	// auto rollback waits, since it has no caller to report ErrDatastoreLocked to.
 	dmutex *sync.Mutex
 
 	// TransactionManager
@@ -116,7 +118,7 @@ func New(ctx context.Context, c *config.DatastoreConfig, sc schema.Client, cc ca
 		syncTreeMutex:    &sync.RWMutex{},
 		taskPool:         pool.NewSharedTaskPool(ctx, runtime.GOMAXPROCS(0)),
 	}
-	ds.transactionManager = types.NewTransactionManager(NewDatastoreRollbackAdapter(ds))
+	ds.transactionManager = types.NewTransactionManager(NewDatastoreRollbackAdapter(ds), types.WithAutoRollbackLock(ds.dmutex))
 
 	// create cache instance if needed
 	// this is a blocking call

pkg/datastore/transaction_lock_test.go

@@ -0,0 +1,149 @@
+package datastore
+
+import (
+	"context"
+	"runtime"
+	"sync"
+	"testing"
+	"time"
+
+	"github.com/sdcio/data-server/mocks/mockcacheclient"
+	"github.com/sdcio/data-server/mocks/mocktarget"
+	"github.com/sdcio/data-server/pkg/config"
+	schemaClient "github.com/sdcio/data-server/pkg/datastore/clients/schema"
+	"github.com/sdcio/data-server/pkg/datastore/types"
+	"github.com/sdcio/data-server/pkg/pool"
+	"github.com/sdcio/data-server/pkg/tree"
+	"github.com/sdcio/data-server/pkg/tree/consts"
+	treetypes "github.com/sdcio/data-server/pkg/tree/types"
+	"github.com/sdcio/data-server/pkg/utils/testhelper"
+	sdcpb "github.com/sdcio/sdc-protos/sdcpb"
+	"github.com/sdcio/sdc-protos/tree_persist"
+	"go.uber.org/mock/gomock"
+)
+
+// newLockTestDatastore returns a Datastore backed by strict mocks: any unexpected
+// cache write or southbound Set fails the test.
+func newLockTestDatastore(t *testing.T) (*Datastore, *mocktarget.MockTarget, *mockcacheclient.MockCacheClientBound) {
+	t.Helper()
+	ctx := context.Background()
+	ctrl := gomock.NewController(t)
+
+	sc, schema, err := testhelper.InitSDCIOSchema()
+	if err != nil {
+		t.Fatal(err)
+	}
+	scb := schemaClient.NewSchemaClientBound(schema, sc)
+
+	ccb := mockcacheclient.NewMockCacheClientBound(ctrl)
+	ccb.EXPECT().
+		IntentGetAll(gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any()).
+		DoAndReturn(func(_ context.Context, _ []string, intentChan chan<- *tree_persist.Intent, errChan chan<- error) {
+			close(intentChan)
+			close(errChan)
+		}).AnyTimes()
+	ccb.EXPECT().IntentGet(gomock.Any(), consts.RunningIntentName).
+		Return(&tree_persist.Intent{IntentName: consts.RunningIntentName}, nil).AnyTimes()
+
+	sbi := mocktarget.NewMockTarget(ctrl)
+
+	pl := pool.NewSharedTaskPool(ctx, runtime.GOMAXPROCS(0))
+	syncTree, err := tree.NewTreeRoot(ctx, tree.NewTreeContext(scb, pl))
+	if err != nil {
+		t.Fatal(err)
+	}
+	if err := syncTree.FinishInsertionPhase(ctx); err != nil {
+		t.Fatal(err)
+	}
+
+	ds := &Datastore{
+		config:        &config.DatastoreConfig{Validation: config.NewValidationConfig(), Name: "test-ds"},
+		syncTreeMutex: &sync.RWMutex{},
+		syncTree:      syncTree,
+		taskPool:      pl,
+		cacheClient:   ccb,
+		sbi:           sbi,
+		dmutex:        &sync.Mutex{},
+		schemaClient:  scb,
+	}
+	ds.transactionManager = types.NewTransactionManager(NewDatastoreRollbackAdapter(ds), types.WithAutoRollbackLock(ds.dmutex))
+	return ds, sbi, ccb
+}
+
+func TestTransactionSet_DryRunNoOpDoesNotPanic(t *testing.T) {
+	ds, _, _ := newLockTestDatastore(t)
+
+	rsp, err := ds.TransactionSet(context.Background(), "x", nil, nil, time.Minute, true)
+	if err != nil {
+		t.Fatal(err)
+	}
+	if len(rsp.GetWarnings()) == 0 {
+		t.Error("expected the no intents warning")
+	}
+	if err := ds.TransactionConfirm(context.Background(), "x"); err == nil {
+		t.Error("dry-run must not register a transaction")
+	}
+}
+
+// A dry-run replace may validate but must neither reach the device nor change the running state.
+func TestTransactionSet_DryRunReplaceDoesNotTouchDevice(t *testing.T) {
+	ctx := context.Background()
+	ds, _, _ := newLockTestDatastore(t) // no Set / IntentModify expectations registered
+
+	replace := types.NewTransactionIntent(consts.ReplaceIntentName, consts.ReplaceValuesPrio)
+	updates, err := treetypes.ExpandAndConvertIntent(ctx, ds.schemaClient, consts.ReplaceIntentName, consts.ReplaceValuesPrio, []*sdcpb.Update{{
+		Path: &sdcpb.Path{},
+		Value: &sdcpb.TypedValue{Value: &sdcpb.TypedValue_JsonVal{
+			JsonVal: []byte(`{"interface":[{"name":"ethernet-1/1","description":"replaced"}]}`),
+		}},
+	}}, time.Now().Unix())
+	if err != nil {
+		t.Fatal(err)
+	}
+	replace.AddUpdates(updates)
+
+	if _, err := ds.TransactionSet(ctx, "x", nil, replace, time.Minute, true); err != nil {
+		t.Fatal(err)
+	}
+}
+
+// Confirm only touches the transaction manager and must work while another set-path holder owns dmutex.
+func TestTransactionConfirm_NotBlockedByDmutex(t *testing.T) {
+	ctx := context.Background()
+	ds, _, _ := newLockTestDatastore(t)
+
+	if _, err := ds.TransactionSet(ctx, "x", nil, nil, time.Minute, false); err != nil {
+		t.Fatal(err)
+	}
+
+	ds.dmutex.Lock()
+	defer ds.dmutex.Unlock()
+	if err := ds.TransactionConfirm(ctx, "x"); err != nil {
+		t.Fatalf("confirm refused while dmutex is held: %v", err)
+	}
+}
+
+// A no-op transaction is registered and must be released by the auto rollback timer if never confirmed.
+func TestTransactionSet_NoOpIsReleasedOnTimeout(t *testing.T) {
+	ctx := context.Background()
+	ds, _, _ := newLockTestDatastore(t)
+
+	if _, err := ds.TransactionSet(ctx, "x", nil, nil, 20*time.Millisecond, false); err != nil {
+		t.Fatal(err)
+	}
+	if _, err := ds.TransactionSet(ctx, "y", nil, nil, time.Minute, false); err == nil {
+		t.Fatal("second transaction must be refused while the first is unconfirmed")
+	}
+
+	deadline := time.Now().Add(2 * time.Second)
+	for {
+		_, err := ds.TransactionSet(ctx, "z", nil, nil, time.Minute, false)
+		if err == nil {
+			return
+		}
+		if time.Now().After(deadline) {
+			t.Fatalf("no-op transaction was never released: %v", err)
+		}
+		time.Sleep(5 * time.Millisecond)
+	}
+}

pkg/datastore/transaction_rpc.go

@@ -26,7 +26,6 @@ import (
 
 var (
 	ErrDatastoreLocked   = errors.New("datastore is locked, other action is ongoing")
-	ErrContextDone       = errors.New("context is closed (done)")
 	ErrValidation        = errors.New("validation error")
 	ErrNoIntentsProvided = errors.New("no intents provided")
 )
@@ -78,8 +77,9 @@ func (d *Datastore) SdcpbTransactionIntentToInternalTI(ctx context.Context, req
 }
 
 // replaceIntent takes a Transaction and treats it as a replaceIntent, replacing the whole device configuration with the content of the given intent.
+// With dryRun set, the intent is only validated; the device and the syncTree stay untouched.
 // returns the warnings as a []string and potential errors that happend during validation / from SBI Set()
-func (d *Datastore) replaceIntent(ctx context.Context, transaction *types.Transaction) ([]string, error) {
+func (d *Datastore) replaceIntent(ctx context.Context, transaction *types.Transaction, dryRun bool) ([]string, error) {
 	log := logger.FromContext(ctx).WithValues("transaction-type", "replace")
 	ctx = logger.IntoContext(ctx, log)
 
@@ -131,6 +131,10 @@ func (d *Datastore) replaceIntent(ctx context.Context, transaction *types.Transa
 	warnings := validationResult.WarningsStr()
 	log.Info("transaction validation passed", "transaction-id", transaction.GetTransactionId(), "stats", validationStats.String())
 
+	if dryRun {
+		return warnings, nil
+	}
+
 	// we use the TargetSourceReplace, that adjustes the tree results in a way
 	// that the whole config tree is getting replaced.
 	replaceRoot := types.NewTargetSourceReplace(adapter.NewEntryOutputAdapter(root.Entry))
@@ -398,14 +402,6 @@ func (d *Datastore) lowlevelTransactionSet(ctx context.Context, transaction *typ
 	}
 
 	log.Info("transaction completed")
-	// start the rollback ticker only if it was not already a rollback transaction.
-	if !transaction.IsRollback() {
-		// start the RollbackTimer
-		err = transaction.StartRollbackTimer(ctx)
-		if err != nil {
-			return nil, err
-		}
-	}
 
 	return result, nil
 }
@@ -480,35 +476,21 @@ func (d *Datastore) TransactionSet(ctx context.Context, transactionId string, tr
 		}
 		defer d.dmutex.Unlock()
 
-		// Try to register the Transaction in the TransactionManager only a single transaction can be register (implicitly being active)
-	outerloop:
-		for {
-			select {
-			case <-ctx.Done():
-				// Context was canceled or timed out
-				log.Error(ctx.Err(), "transaction context canceled or timed out", ctx.Err())
-				return nil, ErrContextDone
-			default:
-				// Start a transaction and prepare to cancel it if any error occurs
-				transactionGuard, err = d.transactionManager.RegisterTransaction(ctx, transaction)
-				if err != nil {
-					return nil, err
-				}
-				if transactionGuard != nil {
-					defer transactionGuard.Done()
-					break outerloop
-				}
-				return nil, ErrDatastoreLocked
-			}
+		// Register the Transaction in the TransactionManager, only a single transaction can be registered (implicitly being active).
+		// If anything fails before the transaction is committed, the guard unregisters it again.
+		transactionGuard, err = d.transactionManager.RegisterTransaction(ctx, transaction)
+		if err != nil {
+			return nil, err
 		}
+		defer transactionGuard.Done()
 	}
 
 	// add the replaceIntent to the transaction
 	transaction.SetReplace(replaceIntent)
 
 	// if replace intent is provided, kickoff the replace intent processing first
 	if transaction.GetReplace() != nil {
-		replaceWarn, err := d.replaceIntent(ctx, transaction)
+		replaceWarn, err := d.replaceIntent(ctx, transaction, dryRun)
 		if err != nil {
 			log.Error(err, "error setting replace intent")
 			return nil, err
@@ -525,8 +507,12 @@ func (d *Datastore) TransactionSet(ctx context.Context, transactionId string, tr
 
 	// no-op transaction
 	if transaction.IsNoOp() {
-		// we expect a transaction confirm from the client
-		transactionGuard.Success()
+		if !dryRun {
+			// we expect a transaction confirm from the client
+			if err := commitTransaction(ctx, transaction, transactionGuard); err != nil {
+				return nil, err
+			}
+		}
 		return &sdcpb.TransactionSetResponse{
 			Warnings: []string{"no intents provided"},
 		}, nil
@@ -547,14 +533,25 @@ func (d *Datastore) TransactionSet(ctx context.Context, transactionId string, tr
 
 	// Mark the transaction as successfully committed
 	if !dryRun {
-		// Mark the transaction as successfully committed
-		transactionGuard.Success()
+		if err := commitTransaction(ctx, transaction, transactionGuard); err != nil {
+			return nil, err
+		}
 
 		log.Info("transaction success")
 	}
 	return response, err
 }
 
+// commitTransaction starts the auto rollback timer of the transaction and keeps it registered,
+// awaiting the client's Confirm. On error the guard still unregisters the transaction.
+func commitTransaction(ctx context.Context, transaction *types.Transaction, guard *types.TransactionGuard) error {
+	if err := transaction.StartRollbackTimer(ctx); err != nil {
+		return err
+	}
+	guard.Success()
+	return nil
+}
+
 func updateToSdcpbUpdate(lvs api.LeafVariantSlice) ([]*sdcpb.Update, error) {
 	result := make([]*sdcpb.Update, 0, len(lvs))
 	for _, lv := range lvs {
@@ -573,11 +570,8 @@ func (d *Datastore) TransactionConfirm(ctx context.Context, transactionId string
 	log := logger.FromContext(ctx)
 	log.Info("transaction confirm")
 
-	if !d.dmutex.TryLock() {
-		return ErrDatastoreLocked
-	}
-	defer d.dmutex.Unlock()
-	// everything remains as is
+	// Confirm never touches the device, so it does not take dmutex and cannot be
+	// refused while Sync Revert or another set-path holder owns it.
 	return d.transactionManager.Confirm(transactionId)
 }
 

pkg/datastore/transaction_rpc_test.go

@@ -179,7 +179,7 @@ func TestTransactionSet_PreviouslyApplied(t *testing.T) {
 				dmutex:        &sync.Mutex{},
 				schemaClient:  scb,
 			}
-			ds.transactionManager = types.NewTransactionManager(NewDatastoreRollbackAdapter(ds))
+			ds.transactionManager = types.NewTransactionManager(NewDatastoreRollbackAdapter(ds), types.WithAutoRollbackLock(ds.dmutex))
 
 			// Prepare Transaction Request
 			transactionId := "txn-1"