[raft/store] Update Transact signature

cmds/db-manager/cleanup/evict.go

@@ -14,6 +14,7 @@ import (
 	scdmodels "github.com/interuss/dss/pkg/scd/models"
 	scdrepos "github.com/interuss/dss/pkg/scd/repos"
 	scds "github.com/interuss/dss/pkg/scd/store"
+	dssstore "github.com/interuss/dss/pkg/store"
 	"github.com/interuss/stacktrace"
 	"github.com/spf13/cobra"
 	"github.com/spf13/pflag"
@@ -100,7 +101,7 @@ func evict(cmd *cobra.Command, _ []string) error {
 		}
 		return nil
 	}
-	if err = scdStore.Transact(ctx, scdAction); err != nil {
+	if _, err = scdStore.Transact(ctx, dssstore.NewFuncOperation(scdAction)); err != nil {
 		return fmt.Errorf("failed to execute SCD transaction: %w", err)
 	}
 
@@ -145,7 +146,7 @@ func evict(cmd *cobra.Command, _ []string) error {
 
 		return nil
 	}
-	if err = ridStore.Transact(ctx, ridAction); err != nil {
+	if _, err = ridStore.Transact(ctx, dssstore.NewFuncOperation(ridAction)); err != nil {
 		return fmt.Errorf("failed to execute RID transaction: %w", err)
 	}
 

pkg/aux_/actions/registry.go

@@ -0,0 +1,10 @@
+package actions
+
+import (
+	"github.com/interuss/dss/pkg/aux_/repos"
+	dssstore "github.com/interuss/dss/pkg/store"
+)
+
+// Registry maps operation IDs to their handlers.
+// TODO: implement
+var Registry = map[string]dssstore.OperationHandler[repos.Repository]{}

pkg/aux_/store/sqlstore/store.go

@@ -3,6 +3,7 @@ package sqlstore
 import (
 	"context"
 
+	"github.com/interuss/dss/pkg/aux_/actions"
 	"github.com/interuss/dss/pkg/aux_/repos"
 	"github.com/interuss/dss/pkg/logging"
 	dssql "github.com/interuss/dss/pkg/sql"
@@ -39,5 +40,6 @@ func Init(ctx context.Context, logger *zap.Logger, withCheckCron bool) (*sqlstor
 				version:   version,
 			}
 		},
+		Registry: actions.Registry,
 	}, withCheckCron)
 }

pkg/memstore/store.go

@@ -12,6 +12,7 @@ import (
 
 	dsserr "github.com/interuss/dss/pkg/errors"
 	"github.com/interuss/dss/pkg/logging"
+	"github.com/interuss/dss/pkg/store"
 	"github.com/interuss/stacktrace"
 	"go.uber.org/zap"
 )
@@ -50,8 +51,8 @@ func Init[R any](ctx context.Context, logger *zap.Logger, name string, r MemRepo
 	return store, nil
 }
 
-func (s *Store[R]) Transact(ctx context.Context, _ func(context.Context, R) error) error {
-	return stacktrace.NewErrorWithCode(dsserr.NotImplemented, "Transact not implemented for memstore")
+func (s *Store[R]) Transact(ctx context.Context, _ store.OperationRequest) (any, error) {
+	return nil, stacktrace.NewErrorWithCode(dsserr.NotImplemented, "Transact not implemented for memstore")
 }
 
 func (s *Store[R]) Interact(_ context.Context) (R, error) {

pkg/raftstore/store.go

@@ -5,6 +5,7 @@ import (
 
 	"github.com/interuss/dss/pkg/raftstore/consensus"
 	raftparams "github.com/interuss/dss/pkg/raftstore/params"
+	"github.com/interuss/dss/pkg/store"
 	"github.com/interuss/stacktrace"
 	"go.uber.org/zap"
 )
@@ -29,9 +30,9 @@ func Init[R any](ctx context.Context, logger *zap.Logger, params raftparams.Conn
 }
 
 // Transact proposes the entry to Raft and blocks until it is committed and applied.
-func (s *Store[R]) Transact(ctx context.Context, f func(context.Context, R) error) error {
+func (s *Store[R]) Transact(_ context.Context, _ store.OperationRequest) (any, error) {
 	// TODO: implement
-	return nil
+	return nil, nil
 }
 
 // Interact returns a repository that can be used to query the store without proposing a Raft entry.

pkg/rid/actions/registry.go

@@ -0,0 +1,10 @@
+package actions
+
+import (
+	"github.com/interuss/dss/pkg/rid/repos"
+	dssstore "github.com/interuss/dss/pkg/store"
+)
+
+// Registry maps operation IDs to their handlers.
+// TODO: implement
+var Registry = map[string]dssstore.OperationHandler[repos.Repository]{}

pkg/rid/application/application_test.go

@@ -13,6 +13,8 @@ import (
 	dssql "github.com/interuss/dss/pkg/sql"
 	"github.com/interuss/dss/pkg/sqlstore"
 	"github.com/interuss/dss/pkg/sqlstore/params"
+	dssstore "github.com/interuss/dss/pkg/store"
+	"github.com/interuss/stacktrace"
 	"github.com/jonboulle/clockwork"
 	"github.com/stretchr/testify/require"
 
@@ -35,8 +37,11 @@ func (s *mockRepo) Interact(ctx context.Context) (repos.Repository, error) {
 	return s, nil
 }
 
-func (s *mockRepo) Transact(ctx context.Context, f func(ctx context.Context, repo repos.Repository) error) error {
-	return f(ctx, s)
+func (s *mockRepo) Transact(ctx context.Context, request dssstore.OperationRequest) (any, error) {
+	if fn, ok := request.(*dssstore.FuncOperation[repos.Repository]); ok {
+		return nil, fn.Execute(ctx, s)
+	}
+	return nil, stacktrace.NewError("unsupported operation type %T", request)
 }
 
 func (s *mockRepo) Close() error {

pkg/rid/application/isa.go

@@ -10,6 +10,7 @@ import (
 	dssmodels "github.com/interuss/dss/pkg/models"
 	ridmodels "github.com/interuss/dss/pkg/rid/models"
 	"github.com/interuss/dss/pkg/rid/repos"
+	"github.com/interuss/dss/pkg/store"
 	"github.com/interuss/stacktrace"
 )
 
@@ -63,7 +64,7 @@ func (a *app) DeleteISA(ctx context.Context, id dssmodels.ID, owner dssmodels.Ow
 		subs []*ridmodels.Subscription
 	)
 	// The following will automatically retry TXN retry errors.
-	err := a.store.Transact(ctx, func(ctx context.Context, repo repos.Repository) error {
+	_, err := a.store.Transact(ctx, store.NewFuncOperation(func(ctx context.Context, repo repos.Repository) error {
 		old, err := repo.GetISA(ctx, id, true)
 		switch {
 		case err != nil:
@@ -88,7 +89,7 @@ func (a *app) DeleteISA(ctx context.Context, id dssmodels.ID, owner dssmodels.Ow
 			return stacktrace.Propagate(err, "Error updating notification indices")
 		}
 		return nil
-	})
+	}))
 	return ret, subs, err // No need to Propagate this error as this stack layer does not add useful information
 }
 
@@ -104,7 +105,7 @@ func (a *app) InsertISA(ctx context.Context, isa *ridmodels.IdentificationServic
 		subs []*ridmodels.Subscription
 	)
 	// The following will automatically retry TXN retry errors.
-	err := a.store.Transact(ctx, func(ctx context.Context, repo repos.Repository) error {
+	_, err := a.store.Transact(ctx, store.NewFuncOperation(func(ctx context.Context, repo repos.Repository) error {
 		// ensure it doesn't exist yet
 		old, err := repo.GetISA(ctx, isa.ID, false)
 		if err != nil {
@@ -126,7 +127,7 @@ func (a *app) InsertISA(ctx context.Context, isa *ridmodels.IdentificationServic
 			return stacktrace.Propagate(err, "Error inserting ISA")
 		}
 		return nil
-	})
+	}))
 	return ret, subs, err // No need to Propagate this error as this stack layer does not add useful information
 }
 
@@ -138,7 +139,7 @@ func (a *app) UpdateISA(ctx context.Context, isa *ridmodels.IdentificationServic
 		subs []*ridmodels.Subscription
 	)
 	// The following will automatically retry TXN retry errors.
-	err := a.store.Transact(ctx, func(ctx context.Context, repo repos.Repository) error {
+	_, err := a.store.Transact(ctx, store.NewFuncOperation(func(ctx context.Context, repo repos.Repository) error {
 		var err error
 
 		old, err := repo.GetISA(ctx, isa.ID, true)
@@ -176,7 +177,7 @@ func (a *app) UpdateISA(ctx context.Context, isa *ridmodels.IdentificationServic
 			return stacktrace.Propagate(err, "Error updating notification indices")
 		}
 		return nil
-	})
+	}))
 
 	return ret, subs, err // No need to Propagate this error as this stack layer does not add useful information
 }

pkg/rid/application/subscription.go

@@ -8,6 +8,7 @@ import (
 	dssmodels "github.com/interuss/dss/pkg/models"
 	ridmodels "github.com/interuss/dss/pkg/rid/models"
 	"github.com/interuss/dss/pkg/rid/repos"
+	"github.com/interuss/dss/pkg/store"
 	"github.com/interuss/stacktrace"
 	"go.uber.org/zap"
 )
@@ -60,7 +61,7 @@ func (a *app) InsertSubscription(ctx context.Context, s *ridmodels.Subscription)
 		return nil, stacktrace.Propagate(err, "Unable to adjust time range")
 	}
 	var sub *ridmodels.Subscription
-	err := a.store.Transact(ctx, func(ctx context.Context, repo repos.Repository) error {
+	_, err := a.store.Transact(ctx, store.NewFuncOperation(func(ctx context.Context, repo repos.Repository) error {
 
 		// ensure it doesn't exist yet
 		old, err := repo.GetSubscription(ctx, s.ID)
@@ -90,15 +91,15 @@ func (a *app) InsertSubscription(ctx context.Context, s *ridmodels.Subscription)
 		}
 
 		return nil
-	})
+	}))
 	return sub, err
 }
 
 // InsertSubscription implements the App InsertSubscription method
 func (a *app) UpdateSubscription(ctx context.Context, s *ridmodels.Subscription) (*ridmodels.Subscription, error) {
 	var sub *ridmodels.Subscription
 
-	err := a.store.Transact(ctx, func(ctx context.Context, repo repos.Repository) error {
+	_, err := a.store.Transact(ctx, store.NewFuncOperation(func(ctx context.Context, repo repos.Repository) error {
 		old, err := repo.GetSubscription(ctx, s.ID)
 		switch {
 		case err != nil:
@@ -138,14 +139,14 @@ func (a *app) UpdateSubscription(ctx context.Context, s *ridmodels.Subscription)
 			return stacktrace.Propagate(err, "Error updating Subscription in repo")
 		}
 		return nil
-	})
+	}))
 	return sub, err
 }
 
 // DeleteSubscription deletes the Subscription identified by "id" and owned by "owner".
 func (a *app) DeleteSubscription(ctx context.Context, id dssmodels.ID, owner dssmodels.Owner, version *dssmodels.Version) (*ridmodels.Subscription, error) {
 	var ret *ridmodels.Subscription
-	err := a.store.Transact(ctx, func(ctx context.Context, repo repos.Repository) error {
+	_, err := a.store.Transact(ctx, store.NewFuncOperation(func(ctx context.Context, repo repos.Repository) error {
 		var err error
 		old, err := repo.GetSubscription(ctx, id)
 		switch {
@@ -168,6 +169,6 @@ func (a *app) DeleteSubscription(ctx context.Context, id dssmodels.ID, owner dss
 			return stacktrace.Propagate(err, "Error deleting Subscription from repo")
 		}
 		return nil
-	})
+	}))
 	return ret, err
 }

pkg/rid/store/sqlstore/store.go

@@ -6,6 +6,7 @@ import (
 	dssql "github.com/interuss/dss/pkg/sql"
 
 	"github.com/interuss/dss/pkg/logging"
+	"github.com/interuss/dss/pkg/rid/actions"
 	"github.com/interuss/dss/pkg/rid/repos"
 	"github.com/interuss/dss/pkg/sqlstore"
 	"github.com/interuss/dss/pkg/store/params"
@@ -44,5 +45,6 @@ func Init(ctx context.Context, logger *zap.Logger, withCheckCron bool) (*sqlstor
 				timeBasedNotificationIndex: opts.TimeBasedNotificationIndex,
 			}
 		},
+		Registry: actions.Registry,
 	}, withCheckCron)
 }

pkg/rid/store/sqlstore/store_test.go

@@ -13,6 +13,7 @@ import (
 	"github.com/interuss/dss/pkg/rid/repos"
 	"github.com/interuss/dss/pkg/sqlstore"
 	"github.com/interuss/dss/pkg/sqlstore/params"
+	dssstore "github.com/interuss/dss/pkg/store"
 	"github.com/jackc/pgx/v5/pgconn"
 	"github.com/jonboulle/clockwork"
 	"github.com/stretchr/testify/require"
@@ -97,12 +98,12 @@ func TestTxnRetrier(t *testing.T) {
 	require.NotNil(t, store)
 	defer tearDownStore()
 
-	err := store.Transact(ctx, func(ctx context.Context, repo repos.Repository) error {
+	_, err := store.Transact(ctx, dssstore.NewFuncOperation(func(ctx context.Context, repo repos.Repository) error {
 		// can query within this
 		isa, err := repo.InsertISA(ctx, serviceArea)
 		require.NotNil(t, isa)
 		return err
-	})
+	}))
 	require.NoError(t, err)
 	// can query afterwads
 	repo, err := store.Interact(ctx)
@@ -118,12 +119,12 @@ func TestTxnRetrier(t *testing.T) {
 	ctx, cancel := context.WithTimeout(ctx, 20*time.Millisecond)
 	defer cancel()
 	count := 0
-	err = store.Transact(ctx, func(ctx context.Context, repo repos.Repository) error {
+	_, err = store.Transact(ctx, dssstore.NewFuncOperation(func(ctx context.Context, repo repos.Repository) error {
 		// can query within this
 		count++
 		// Postgre retryable error
 		return &pgconn.PgError{Code: "40001"}
-	})
+	}))
 	require.Error(t, err)
 	// Ensure it was retried.
 	require.Greater(t, count, 1)
@@ -141,13 +142,13 @@ func TestTransactor(t *testing.T) {
 	subscription2 := subscriptionsPool[1].input
 
 	txnCount := 0
-	err := store.Transact(ctx, func(ctx context.Context, s1 repos.Repository) error {
+	_, err := store.Transact(ctx, dssstore.NewFuncOperation(func(ctx context.Context, s1 repos.Repository) error {
 		// We should get to this retry, then return nothing.
 		if txnCount > 0 {
 			return errors.New("already failed")
 		}
 		txnCount++
-		err := store.Transact(ctx, func(ctx context.Context, s2 repos.Repository) error {
+		_, err := store.Transact(ctx, dssstore.NewFuncOperation(func(ctx context.Context, s2 repos.Repository) error {
 			subs, err := s1.SearchSubscriptions(ctx, subscription1.Cells)
 			require.NoError(t, err)
 			require.Len(t, subs, 0)
@@ -164,9 +165,9 @@ func TestTransactor(t *testing.T) {
 			require.NoError(t, err)
 
 			return nil
-		})
+		}))
 		return err
-	})
+	}))
 	require.Error(t, err)
 
 	repo, err := store.Interact(ctx)

pkg/scd/actions/registry.go

@@ -0,0 +1,10 @@
+package actions
+
+import (
+	"github.com/interuss/dss/pkg/scd/repos"
+	dssstore "github.com/interuss/dss/pkg/store"
+)
+
+// Registry maps operation IDs to their handlers
+// TODO: implement
+var Registry = map[string]dssstore.OperationHandler[repos.Repository]{}

pkg/scd/constraints_handler.go

@@ -11,6 +11,7 @@ import (
 	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"
 )
@@ -88,7 +89,7 @@ func (a *Server) DeleteConstraintReference(ctx context.Context, req *restapi.Del
 		return nil
 	}
 
-	err = a.Store.Transact(ctx, action)
+	_, err = a.Store.Transact(ctx, dssstore.NewFuncOperation(action))
 	if err != nil {
 		err = stacktrace.Propagate(err, "Could not delete constraint")
 		errResp := &restapi.ErrorResponse{Message: dsserr.Handle(ctx, err)}
@@ -147,7 +148,7 @@ func (a *Server) GetConstraintReference(ctx context.Context, req *restapi.GetCon
 		return nil
 	}
 
-	err = a.Store.Transact(ctx, action)
+	_, err = a.Store.Transact(ctx, dssstore.NewFuncOperation(action))
 	if err != nil {
 		err = stacktrace.Propagate(err, "Could not get constraint")
 		if stacktrace.GetCode(err) == dsserr.NotFound {
@@ -306,7 +307,7 @@ func (a *Server) PutConstraintReference(ctx context.Context, manager string, ent
 		return nil
 	}
 
-	err = a.Store.Transact(ctx, action)
+	_, 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
 	}
@@ -434,7 +435,7 @@ func (a *Server) QueryConstraintReferences(ctx context.Context, req *restapi.Que
 		return nil
 	}
 
-	err = a.Store.Transact(ctx, action)
+	_, err = a.Store.Transact(ctx, dssstore.NewFuncOperation(action))
 	if err != nil {
 		return restapi.QueryConstraintReferencesResponseSet{Response500: &api.InternalServerErrorBody{
 			ErrorMessage: *dsserr.Handle(ctx, stacktrace.Propagate(err, "Got an unexpected error"))}}