From 55862c632c36a0cc45d61b6566d6541671d0dd8d Mon Sep 17 00:00:00 2001 From: Mariem Baccari Date: Mon, 29 Jun 2026 12:02:46 +0200 Subject: [PATCH 1/3] [raft/store] Update Transact signature --- cmds/db-manager/cleanup/evict.go | 5 +-- pkg/aux_/actions/registry.go | 10 ++++++ pkg/aux_/store/sqlstore/store.go | 2 ++ pkg/memstore/store.go | 5 +-- pkg/raftstore/store.go | 5 +-- pkg/rid/actions/registry.go | 10 ++++++ pkg/rid/application/application_test.go | 9 +++-- pkg/rid/application/isa.go | 13 ++++---- pkg/rid/application/subscription.go | 13 ++++---- pkg/rid/store/sqlstore/store.go | 2 ++ pkg/rid/store/sqlstore/store_test.go | 17 +++++----- pkg/scd/actions/registry.go | 10 ++++++ pkg/scd/constraints_handler.go | 9 ++--- pkg/scd/operational_intents_handler.go | 9 ++--- pkg/scd/store/sqlstore/store.go | 2 ++ pkg/scd/subscriptions_handler.go | 9 ++--- pkg/scd/uss_availability_handler.go | 5 +-- pkg/sqlstore/store.go | 39 +++++++++++++++++----- pkg/store/store.go | 44 +++++++++++++++++++++++-- 19 files changed, 164 insertions(+), 54 deletions(-) create mode 100644 pkg/aux_/actions/registry.go create mode 100644 pkg/rid/actions/registry.go create mode 100644 pkg/scd/actions/registry.go diff --git a/cmds/db-manager/cleanup/evict.go b/cmds/db-manager/cleanup/evict.go index 572b90f5b..6cdf70d62 100644 --- a/cmds/db-manager/cleanup/evict.go +++ b/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) } diff --git a/pkg/aux_/actions/registry.go b/pkg/aux_/actions/registry.go new file mode 100644 index 000000000..d16395b88 --- /dev/null +++ b/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]{} diff --git a/pkg/aux_/store/sqlstore/store.go b/pkg/aux_/store/sqlstore/store.go index 11c5f5e7a..772c3e4fd 100644 --- a/pkg/aux_/store/sqlstore/store.go +++ b/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) } diff --git a/pkg/memstore/store.go b/pkg/memstore/store.go index 9c850f2c3..026e3c2c5 100644 --- a/pkg/memstore/store.go +++ b/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) { diff --git a/pkg/raftstore/store.go b/pkg/raftstore/store.go index ee2b40b51..12bdb78bd 100644 --- a/pkg/raftstore/store.go +++ b/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. diff --git a/pkg/rid/actions/registry.go b/pkg/rid/actions/registry.go new file mode 100644 index 000000000..82c125a55 --- /dev/null +++ b/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]{} diff --git a/pkg/rid/application/application_test.go b/pkg/rid/application/application_test.go index 0cc3fd885..7ac8db967 100644 --- a/pkg/rid/application/application_test.go +++ b/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 { diff --git a/pkg/rid/application/isa.go b/pkg/rid/application/isa.go index ea402af60..a84170a4f 100644 --- a/pkg/rid/application/isa.go +++ b/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 } diff --git a/pkg/rid/application/subscription.go b/pkg/rid/application/subscription.go index 33ba2af36..2e666adeb 100644 --- a/pkg/rid/application/subscription.go +++ b/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,7 +91,7 @@ func (a *app) InsertSubscription(ctx context.Context, s *ridmodels.Subscription) } return nil - }) + })) return sub, err } @@ -98,7 +99,7 @@ func (a *app) InsertSubscription(ctx context.Context, s *ridmodels.Subscription) 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 } diff --git a/pkg/rid/store/sqlstore/store.go b/pkg/rid/store/sqlstore/store.go index 90bb20325..20bb2e026 100644 --- a/pkg/rid/store/sqlstore/store.go +++ b/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) } diff --git a/pkg/rid/store/sqlstore/store_test.go b/pkg/rid/store/sqlstore/store_test.go index 275a1a332..596a1f958 100644 --- a/pkg/rid/store/sqlstore/store_test.go +++ b/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) diff --git a/pkg/scd/actions/registry.go b/pkg/scd/actions/registry.go new file mode 100644 index 000000000..c8389cc26 --- /dev/null +++ b/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]{} diff --git a/pkg/scd/constraints_handler.go b/pkg/scd/constraints_handler.go index 0688176c7..a53c23a60 100644 --- a/pkg/scd/constraints_handler.go +++ b/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"))}} diff --git a/pkg/scd/operational_intents_handler.go b/pkg/scd/operational_intents_handler.go index f611797d0..b6446dbaf 100644 --- a/pkg/scd/operational_intents_handler.go +++ b/pkg/scd/operational_intents_handler.go @@ -13,6 +13,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" ) @@ -165,7 +166,7 @@ func (a *Server) DeleteOperationalIntentReference(ctx context.Context, req *rest 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 operational intent") errResp := &restapi.ErrorResponse{Message: dsserr.Handle(ctx, err)} @@ -221,7 +222,7 @@ func (a *Server) GetOperationalIntentReference(ctx context.Context, req *restapi 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 operational intent") if stacktrace.GetCode(err) == dsserr.NotFound { @@ -288,7 +289,7 @@ func (a *Server) QueryOperationalIntentReferences(ctx context.Context, req *rest 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 query operational intent") if stacktrace.GetCode(err) == dsserr.BadRequest { @@ -934,7 +935,7 @@ func (a *Server) upsertOperationalIntentReference(ctx context.Context, now time. return nil } - err = a.Store.Transact(ctx, action) + _, err = a.Store.Transact(ctx, dssstore.NewFuncOperation(action)) if err != nil { return nil, responseConflict, err // No need to Propagate this error as this is not a useful stacktrace line } diff --git a/pkg/scd/store/sqlstore/store.go b/pkg/scd/store/sqlstore/store.go index 54b509720..3c2bb8f54 100644 --- a/pkg/scd/store/sqlstore/store.go +++ b/pkg/scd/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/scd/actions" "github.com/interuss/dss/pkg/scd/repos" "github.com/interuss/dss/pkg/sqlstore" "github.com/interuss/dss/pkg/store/params" @@ -46,5 +47,6 @@ func Init(ctx context.Context, logger *zap.Logger, withCheckCron bool) (*sqlstor timeBasedNotificationIndex: opts.TimeBasedNotificationIndex, } }, + Registry: actions.Registry, }, withCheckCron) } diff --git a/pkg/scd/subscriptions_handler.go b/pkg/scd/subscriptions_handler.go index 98ab9fed2..3aec27e5b 100644 --- a/pkg/scd/subscriptions_handler.go +++ b/pkg/scd/subscriptions_handler.go @@ -12,6 +12,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/jonboulle/clockwork" ) @@ -276,7 +277,7 @@ func (a *Server) PutSubscription(ctx context.Context, manager string, subscripti 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 } @@ -340,7 +341,7 @@ func (a *Server) GetSubscription(ctx context.Context, req *restapi.GetSubscripti 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 subscription") errResp := &restapi.ErrorResponse{Message: dsserr.Handle(ctx, err)} @@ -424,7 +425,7 @@ func (a *Server) QuerySubscriptions(ctx context.Context, req *restapi.QuerySubsc return nil } - err = a.Store.Transact(ctx, action) + _, err = a.Store.Transact(ctx, dssstore.NewFuncOperation(action)) if err != nil { errResp := &restapi.ErrorResponse{Message: dsserr.Handle(ctx, err)} @@ -517,7 +518,7 @@ func (a *Server) DeleteSubscription(ctx context.Context, req *restapi.DeleteSubs 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 subscription") errResp := &restapi.ErrorResponse{Message: dsserr.Handle(ctx, err)} diff --git a/pkg/scd/uss_availability_handler.go b/pkg/scd/uss_availability_handler.go index 517a9859a..607876afc 100644 --- a/pkg/scd/uss_availability_handler.go +++ b/pkg/scd/uss_availability_handler.go @@ -10,6 +10,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" ) @@ -51,7 +52,7 @@ func (a *Server) GetUssAvailability(ctx context.Context, req *restapi.GetUssAvai return nil } - err := a.Store.Transact(ctx, action) + _, err := a.Store.Transact(ctx, dssstore.NewFuncOperation(action)) if err != nil { // In case of older DB versions where availability table doesn't exist if strings.Contains(err.Error(), "does not exist") { @@ -121,7 +122,7 @@ func (a *Server) SetUssAvailability(ctx context.Context, req *restapi.SetUssAvai } return nil } - err = a.Store.Transact(ctx, action) + _, err = a.Store.Transact(ctx, dssstore.NewFuncOperation(action)) if err != nil { // In case of older DB versions where availability table doesn't exist if strings.Contains(err.Error(), "does not exist") { diff --git a/pkg/sqlstore/store.go b/pkg/sqlstore/store.go index 92437c190..753d39b2e 100644 --- a/pkg/sqlstore/store.go +++ b/pkg/sqlstore/store.go @@ -35,6 +35,7 @@ type Store[R any] struct { Clock clockwork.Clock schemaVersion *semver.Version newRepo func(dsssql.Queryable) R + registry map[string]store.OperationHandler[R] maxRetries int } @@ -45,6 +46,7 @@ type Config[R any] struct { CrdbMajorSchemaVersion int64 YbMajorSchemaVersion int64 NewRepo func(q dsssql.Queryable, clock clockwork.Clock, v *Version) R + Registry map[string]store.OperationHandler[R] } func checkMajorSchemaVersion[R any](ctx context.Context, db *Store[R], vs *semver.Version, crdbExpected int64, ybExpected int64) error { @@ -155,6 +157,7 @@ func Init[R any](ctx context.Context, cfg Config[R], withCheckCron bool) (*Store db.newRepo = func(q dsssql.Queryable) R { return cfg.NewRepo(q, db.Clock, db.Version) } + db.registry = cfg.Registry if withCheckCron { c := cron.New() @@ -176,25 +179,31 @@ func (s *Store[R]) Interact(_ context.Context) (R, error) { return s.newRepo(s.Pool), nil } -func (s *Store[R]) Transact(ctx context.Context, f func(context.Context, R) error) error { +func (s *Store[R]) Transact(ctx context.Context, request store.OperationRequest) (any, error) { if s.Version.Type == Yugabyte { - return s.transactYugabyte(ctx, f) + return s.transactYugabyte(ctx, request) } ctx = crdb.WithMaxRetries(ctx, s.maxRetries) - return crdbpgx.ExecuteTx(ctx, s.Pool, pgx.TxOptions{IsoLevel: pgx.Serializable}, func(tx pgx.Tx) error { - return f(ctx, s.newRepo(tx)) + var result any + var err error + err = crdbpgx.ExecuteTx(ctx, s.Pool, pgx.TxOptions{IsoLevel: pgx.Serializable}, func(tx pgx.Tx) error { + result, err = s.execute(ctx, s.newRepo(tx), request) + return err }) + return result, err } -func (s *Store[R]) transactYugabyte(ctx context.Context, f func(context.Context, R) error) error { +func (s *Store[R]) transactYugabyte(ctx context.Context, request store.OperationRequest) (any, error) { + var result any var err error for attempt := 0; attempt <= s.maxRetries; attempt++ { err = pgx.BeginTxFunc(ctx, s.Pool, pgx.TxOptions{IsoLevel: pgx.Serializable}, func(tx pgx.Tx) error { - return f(ctx, s.newRepo(tx)) + result, err = s.execute(ctx, s.newRepo(tx), request) + return err }) if err == nil || !isYugabyteRetryable(err) { - return err + return result, err } if attempt == s.maxRetries { break @@ -202,11 +211,23 @@ func (s *Store[R]) transactYugabyte(ctx context.Context, f func(context.Context, backoff := time.Duration(1< Date: Wed, 8 Jul 2026 15:31:45 +0200 Subject: [PATCH 2/3] [raft/store] Add IsReadOnly to handler --- pkg/store/store.go | 7 ++++--- 1 file changed, 4 insertions(+), 3 deletions(-) diff --git a/pkg/store/store.go b/pkg/store/store.go index 6cb8448c1..86e8275f7 100644 --- a/pkg/store/store.go +++ b/pkg/store/store.go @@ -25,9 +25,10 @@ type OperationRequest interface { // OperationHandler holds the logic for the encoding and execution of a registered operation type OperationHandler[R any] struct { - Encode func(req OperationRequest) ([]byte, error) - Decode func(buf []byte) (OperationRequest, error) - Execute func(ctx context.Context, repo R, request OperationRequest) (any, error) + Encode func(req OperationRequest) ([]byte, error) + Decode func(buf []byte) (OperationRequest, error) + Execute func(ctx context.Context, repo R, request OperationRequest) (any, error) + IsReadOnly bool } // TransactWithResult wraps Store.Transact and casts the result to ResultType, avoiding a cast at every call site. From aaae8fd53ebb46fec5585f34ec508a565908d5cd Mon Sep 17 00:00:00 2001 From: Mariem Baccari Date: Fri, 22 May 2026 09:23:39 +0200 Subject: [PATCH 3/3] [raft/store] Add generic store --- pkg/aux_/store/raftstore/store.go | 19 +++++- pkg/raftstore/consensus/consensus.go | 2 +- pkg/raftstore/consensus/proposal.go | 13 ++-- pkg/raftstore/store.go | 95 +++++++++++++++++++++++----- pkg/rid/store/raftstore/store.go | 19 +++++- pkg/scd/store/raftstore/store.go | 19 +++++- 6 files changed, 140 insertions(+), 27 deletions(-) diff --git a/pkg/aux_/store/raftstore/store.go b/pkg/aux_/store/raftstore/store.go index d09162d87..f74c1ccf0 100644 --- a/pkg/aux_/store/raftstore/store.go +++ b/pkg/aux_/store/raftstore/store.go @@ -3,9 +3,12 @@ package raftstore import ( "context" + "github.com/interuss/dss/pkg/aux_/actions" "github.com/interuss/dss/pkg/aux_/repos" auxraftparams "github.com/interuss/dss/pkg/aux_/store/raftstore/params" + dsserr "github.com/interuss/dss/pkg/errors" "github.com/interuss/dss/pkg/raftstore" + "github.com/interuss/dss/pkg/raftstore/consensus" "github.com/interuss/stacktrace" "go.uber.org/zap" ) @@ -18,5 +21,19 @@ func Init(ctx context.Context, logger *zap.Logger) (*raftstore.Store[repos.Repos if err != nil { return nil, stacktrace.Propagate(err, "failed to get aux raft parameters") } - return raftstore.Init(ctx, logger.With(zap.String("service", "aux_")), params, func() repos.Repository { return &repo{} }) + return raftstore.Init(ctx, logger.With(zap.String("service", "aux_")), params, &repo{}, actions.Registry) +} + +func (r *repo) GetRepo() repos.Repository { return r } + +func (r *repo) GetSnapshot() ([]byte, error) { + return nil, stacktrace.NewErrorWithCode(dsserr.NotImplemented, "not implemented yet") +} + +func (r *repo) RestoreFromSnapshot([]byte) error { + return stacktrace.NewErrorWithCode(dsserr.NotImplemented, "not implemented yet") +} + +func (r *repo) Apply(_ context.Context, _ consensus.Proposal) (any, error) { + return nil, stacktrace.NewErrorWithCode(dsserr.NotImplemented, "not implemented yet") } diff --git a/pkg/raftstore/consensus/consensus.go b/pkg/raftstore/consensus/consensus.go index f41b36e14..99b8162bb 100644 --- a/pkg/raftstore/consensus/consensus.go +++ b/pkg/raftstore/consensus/consensus.go @@ -131,7 +131,7 @@ func (c *Consensus) Stop(ctx context.Context) { } // HandleClientRequest blocks until the proposal is committed and applied / dropped or until ctx is cancelled. -func (c *Consensus) HandleClientRequest(ctx context.Context, requestType string, value any, readOnly bool) (any, error) { +func (c *Consensus) HandleClientRequest(ctx context.Context, requestType string, value []byte, readOnly bool) (any, error) { proposal, err := c.newProposal(ctx, requestType, value, readOnly) if err != nil { return nil, stacktrace.Propagate(err, "failed to create proposal") diff --git a/pkg/raftstore/consensus/proposal.go b/pkg/raftstore/consensus/proposal.go index a5e837d82..2e550da71 100644 --- a/pkg/raftstore/consensus/proposal.go +++ b/pkg/raftstore/consensus/proposal.go @@ -2,11 +2,11 @@ package consensus import ( "context" - "encoding/json" "sync" "time" "github.com/google/uuid" + "github.com/interuss/dss/pkg/timestamp" "github.com/interuss/stacktrace" ) @@ -30,17 +30,16 @@ type Proposal struct { ReadOnly bool `json:"read_only"` } -func (c *Consensus) newProposal(ctx context.Context, requestType string, payload any, readOnly bool) (Proposal, error) { - // TODO - Fetch timestamp from context - value, err := json.Marshal(payload) - if err != nil { - return Proposal{}, stacktrace.Propagate(err, "failed to serialize proposal payload") +func (c *Consensus) newProposal(ctx context.Context, requestType string, value []byte, readOnly bool) (Proposal, error) { + timestamp, err := timestamp.RequestTimestampFromContext(ctx) + if err != nil || timestamp.IsZero() { + return Proposal{}, stacktrace.Propagate(err, "failed to get timestamp from context") } return Proposal{ ID: uuid.NewString(), NodeID: c.nodeID, - Timestamp: time.Now().UTC(), + Timestamp: timestamp, RequestType: requestType, Value: value, ReadOnly: readOnly, diff --git a/pkg/raftstore/store.go b/pkg/raftstore/store.go index 12bdb78bd..e14aaa668 100644 --- a/pkg/raftstore/store.go +++ b/pkg/raftstore/store.go @@ -2,46 +2,109 @@ package raftstore import ( "context" + "sync" + "github.com/interuss/dss/pkg/logging" "github.com/interuss/dss/pkg/raftstore/consensus" raftparams "github.com/interuss/dss/pkg/raftstore/params" "github.com/interuss/dss/pkg/store" + "github.com/interuss/dss/pkg/timestamp" "github.com/interuss/stacktrace" "go.uber.org/zap" ) +type RaftRepo[R any] interface { + GetRepo() R + // Apply is called on every committed entry. The proposal must be applied atomically. + Apply(ctx context.Context, proposal consensus.Proposal) (any, error) + GetSnapshot() ([]byte, error) + RestoreFromSnapshot(data []byte) error +} + type Store[R any] struct { - newRepo func() R - consensus *consensus.Consensus + logger *zap.Logger + + raftRepo RaftRepo[R] + cancel context.CancelFunc + registry map[string]store.OperationHandler[R] + + Consensus *consensus.Consensus + + wg sync.WaitGroup } -func Init[R any](ctx context.Context, logger *zap.Logger, params raftparams.ConnectParameters, newRepo func() R) (*Store[R], error) { +func Init[R any](ctx context.Context, logger *zap.Logger, params raftparams.ConnectParameters, r RaftRepo[R], registry map[string]store.OperationHandler[R]) (*Store[R], error) { + ctx, cancel := context.WithCancel(ctx) + + store := &Store[R]{ + raftRepo: r, + logger: logging.WithValuesFromContext(ctx, logger), + cancel: cancel, + registry: registry, + } commitC := make(chan consensus.EntryCommit) - consensusInstance, err := consensus.NewConsensus(ctx, logger, params, func() ([]byte, error) { return nil, nil }, commitC) + store.wg.Go(func() { store.processCommits(ctx, commitC) }) + + consensusInstance, err := consensus.NewConsensus(ctx, logger, params, r.GetSnapshot, commitC) if err != nil { return nil, stacktrace.Propagate(err, "failed to initialize consensus") } - // TODO: start consumer goroutine reading from commitC - return &Store[R]{ - newRepo: newRepo, - consensus: consensusInstance, - }, nil + store.Consensus = consensusInstance + + return store, nil } // Transact proposes the entry to Raft and blocks until it is committed and applied. -func (s *Store[R]) Transact(_ context.Context, _ store.OperationRequest) (any, error) { - // TODO: implement - return nil, nil +func (s *Store[R]) Transact(ctx context.Context, request store.OperationRequest) (any, error) { + handler, ok := s.registry[request.OperationID()] + if !ok { + return nil, stacktrace.NewError("no handler registered for operation %q", request.OperationID()) + } + payload, err := handler.Encode(request) + if err != nil { + return nil, stacktrace.Propagate(err, "failed to encode op %q", request.OperationID()) + } + return s.Consensus.HandleClientRequest(ctx, request.OperationID(), payload, handler.IsReadOnly) } -// Interact returns a repository that can be used to query the store without proposing a Raft entry. +// Interact returns the underlying Raft repo which, for every operation, will propose it to Raft and return the results. func (s *Store[R]) Interact(_ context.Context) (R, error) { - return s.newRepo(), nil + return s.raftRepo.GetRepo(), nil } -// Close shuts down the consensus instance. +// Close shuts down the consensus instance and processCommits loop. func (s *Store[R]) Close() error { - // TODO: implement + s.Consensus.Stop(context.Background()) + s.cancel() + s.logger.Info("waiting for commit processing goroutine to exit") + s.wg.Wait() return nil } + +// processCommits reads committed entries from the consensus layer and applies them via Apply. +func (s *Store[R]) processCommits(ctx context.Context, commitCh <-chan consensus.EntryCommit) { + for { + select { + case <-ctx.Done(): + s.logger.Info("stopping commit processing loop") + return + case commit, ok := <-commitCh: + if !ok { + s.logger.Info("commit channel closed, stopping commit processing loop") + return + } + + if commit.SnapshotData != nil { + if err := s.raftRepo.RestoreFromSnapshot(commit.SnapshotData); err != nil { + s.logger.Fatal("failed to restore from snapshot", zap.Error(err)) + } + continue + } + + proposalCtx := timestamp.WithRequestTimestamp(ctx, commit.Prop.Timestamp) + result, err := s.raftRepo.Apply(proposalCtx, commit.Prop) + commit.Done <- consensus.ProposalResult{Result: result, Error: err} + } + } +} diff --git a/pkg/rid/store/raftstore/store.go b/pkg/rid/store/raftstore/store.go index 578aabeac..bc55a3dca 100644 --- a/pkg/rid/store/raftstore/store.go +++ b/pkg/rid/store/raftstore/store.go @@ -3,7 +3,10 @@ package raftstore import ( "context" + dsserr "github.com/interuss/dss/pkg/errors" "github.com/interuss/dss/pkg/raftstore" + "github.com/interuss/dss/pkg/raftstore/consensus" + "github.com/interuss/dss/pkg/rid/actions" "github.com/interuss/dss/pkg/rid/repos" ridraftparams "github.com/interuss/dss/pkg/rid/store/raftstore/params" "github.com/interuss/stacktrace" @@ -18,5 +21,19 @@ func Init(ctx context.Context, logger *zap.Logger) (*raftstore.Store[repos.Repos if err != nil { return nil, stacktrace.Propagate(err, "failed to get rid raft parameters") } - return raftstore.Init(ctx, logger.With(zap.String("service", "rid")), params, func() repos.Repository { return &repo{} }) + return raftstore.Init(ctx, logger.With(zap.String("service", "rid")), params, &repo{}, actions.Registry) +} + +func (r *repo) GetRepo() repos.Repository { return r } + +func (r *repo) GetSnapshot() ([]byte, error) { + return nil, stacktrace.NewErrorWithCode(dsserr.NotImplemented, "not implemented yet") +} + +func (r *repo) RestoreFromSnapshot([]byte) error { + return stacktrace.NewErrorWithCode(dsserr.NotImplemented, "not implemented yet") +} + +func (r *repo) Apply(_ context.Context, _ consensus.Proposal) (any, error) { + return nil, stacktrace.NewErrorWithCode(dsserr.NotImplemented, "not implemented yet") } diff --git a/pkg/scd/store/raftstore/store.go b/pkg/scd/store/raftstore/store.go index a7eeb663a..799d6f1dd 100644 --- a/pkg/scd/store/raftstore/store.go +++ b/pkg/scd/store/raftstore/store.go @@ -3,7 +3,10 @@ package raftstore import ( "context" + dsserr "github.com/interuss/dss/pkg/errors" "github.com/interuss/dss/pkg/raftstore" + "github.com/interuss/dss/pkg/raftstore/consensus" + "github.com/interuss/dss/pkg/scd/actions" "github.com/interuss/dss/pkg/scd/repos" scdraftparams "github.com/interuss/dss/pkg/scd/store/raftstore/params" "github.com/interuss/stacktrace" @@ -18,5 +21,19 @@ func Init(ctx context.Context, logger *zap.Logger) (*raftstore.Store[repos.Repos if err != nil { return nil, stacktrace.Propagate(err, "failed to get scd raft parameters") } - return raftstore.Init(ctx, logger.With(zap.String("service", "scd")), params, func() repos.Repository { return &repo{} }) + return raftstore.Init(ctx, logger.With(zap.String("service", "scd")), params, &repo{}, actions.Registry) +} + +func (r *repo) GetRepo() repos.Repository { return r } + +func (r *repo) GetSnapshot() ([]byte, error) { + return nil, stacktrace.NewErrorWithCode(dsserr.NotImplemented, "not implemented yet") +} + +func (r *repo) RestoreFromSnapshot([]byte) error { + return stacktrace.NewErrorWithCode(dsserr.NotImplemented, "not implemented yet") +} + +func (r *repo) Apply(_ context.Context, _ consensus.Proposal) (any, error) { + return nil, stacktrace.NewErrorWithCode(dsserr.NotImplemented, "not implemented yet") }