Skip to content

[raft] Change transact signature #1574

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Closed
MariemBaccari wants to merge 1 commit into from
Closed

[raft] Change transact signature #1574

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
5 changes: 3 additions & 2 deletions cmds/db-manager/cleanup/evict.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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"
"github.com/interuss/dss/pkg/store"
"github.com/interuss/stacktrace"
"github.com/spf13/cobra"
"github.com/spf13/pflag"
Expand Down Expand Up @@ -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, store.Request{}, scdAction); err != nil {
return fmt.Errorf("failed to execute SCD transaction: %w", err)
}

Expand Down Expand Up @@ -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, store.Request{}, ridAction); err != nil {
return fmt.Errorf("failed to execute RID transaction: %w", err)
}

Expand Down
5 changes: 3 additions & 2 deletions pkg/raftstore/store.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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"
)
Expand All @@ -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(ctx context.Context, _ store.Request, f func(context.Context, R) error) (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.
Expand Down
5 changes: 3 additions & 2 deletions pkg/rid/application/application_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -13,6 +13,7 @@ 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/jonboulle/clockwork"
"github.com/stretchr/testify/require"

Expand All @@ -35,8 +36,8 @@ 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, _ dssstore.Request, f func(ctx context.Context, repo repos.Repository) error) (any, error) {
return nil, f(ctx, s)
}

func (s *mockRepo) Close() error {
Expand Down
7 changes: 4 additions & 3 deletions pkg/rid/application/isa.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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"
)

Expand Down Expand Up @@ -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.Request{}, func(ctx context.Context, repo repos.Repository) error {
old, err := repo.GetISA(ctx, id, true)
switch {
case err != nil:
Expand Down Expand Up @@ -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.Request{}, 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 {
Expand Down Expand Up @@ -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.Request{}, func(ctx context.Context, repo repos.Repository) error {
var err error

old, err := repo.GetISA(ctx, isa.ID, true)
Expand Down
7 changes: 4 additions & 3 deletions pkg/rid/application/subscription.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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"
)
Expand Down Expand Up @@ -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.Request{}, func(ctx context.Context, repo repos.Repository) error {

// ensure it doesn't exist yet
old, err := repo.GetSubscription(ctx, s.ID)
Expand Down Expand Up @@ -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.Request{}, func(ctx context.Context, repo repos.Repository) error {
old, err := repo.GetSubscription(ctx, s.ID)
switch {
case err != nil:
Expand Down Expand Up @@ -145,7 +146,7 @@ func (a *app) UpdateSubscription(ctx context.Context, s *ridmodels.Subscription)
// 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.Request{}, func(ctx context.Context, repo repos.Repository) error {
var err error
old, err := repo.GetSubscription(ctx, id)
switch {
Expand Down
9 changes: 5 additions & 4 deletions pkg/rid/store/sqlstore/store_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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"
Expand Down Expand Up @@ -97,7 +98,7 @@ 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.Request{}, func(ctx context.Context, repo repos.Repository) error {
// can query within this
isa, err := repo.InsertISA(ctx, serviceArea)
require.NotNil(t, isa)
Expand All @@ -118,7 +119,7 @@ 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.Request{}, func(ctx context.Context, repo repos.Repository) error {
// can query within this
count++
// Postgre retryable error
Expand All @@ -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.Request{}, 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.Request{}, func(ctx context.Context, s2 repos.Repository) error {
subs, err := s1.SearchSubscriptions(ctx, subscription1.Cells)
require.NoError(t, err)
require.Len(t, subs, 0)
Expand Down
9 changes: 5 additions & 4 deletions pkg/scd/constraints_handler.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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"
"github.com/interuss/dss/pkg/store"
"github.com/interuss/stacktrace"
"github.com/jackc/pgx/v5"
)
Expand Down Expand Up @@ -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, store.Request{}, action)
if err != nil {
err = stacktrace.Propagate(err, "Could not delete constraint")
errResp := &restapi.ErrorResponse{Message: dsserr.Handle(ctx, err)}
Expand Down Expand Up @@ -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, store.Request{}, action)
if err != nil {
err = stacktrace.Propagate(err, "Could not get constraint")
if stacktrace.GetCode(err) == dsserr.NotFound {
Expand Down Expand Up @@ -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, store.Request{}, action)
if err != nil {
return nil, err // No need to Propagate this error as this is not a useful stacktrace line
}
Expand Down Expand Up @@ -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, store.Request{}, action)
if err != nil {
return restapi.QueryConstraintReferencesResponseSet{Response500: &api.InternalServerErrorBody{
ErrorMessage: *dsserr.Handle(ctx, stacktrace.Propagate(err, "Got an unexpected error"))}}
Expand Down
9 changes: 5 additions & 4 deletions pkg/scd/operational_intents_handler.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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"
"github.com/interuss/dss/pkg/store"
"github.com/interuss/stacktrace"
)

Expand Down Expand Up @@ -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, store.Request{}, action)
if err != nil {
err = stacktrace.Propagate(err, "Could not delete operational intent")
errResp := &restapi.ErrorResponse{Message: dsserr.Handle(ctx, err)}
Expand Down Expand Up @@ -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, store.Request{}, action)
if err != nil {
err = stacktrace.Propagate(err, "Could not get operational intent")
if stacktrace.GetCode(err) == dsserr.NotFound {
Expand Down Expand Up @@ -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, store.Request{}, action)
if err != nil {
err = stacktrace.Propagate(err, "Could not query operational intent")
if stacktrace.GetCode(err) == dsserr.BadRequest {
Expand Down Expand Up @@ -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, store.Request{}, action)
if err != nil {
return nil, responseConflict, err // No need to Propagate this error as this is not a useful stacktrace line
}
Expand Down
9 changes: 5 additions & 4 deletions pkg/scd/subscriptions_handler.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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"
"github.com/interuss/dss/pkg/store"
"github.com/interuss/stacktrace"
"github.com/jonboulle/clockwork"
)
Expand Down Expand Up @@ -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, store.Request{}, action)
if err != nil {
return nil, err // No need to Propagate this error as this is not a useful stacktrace line
}
Expand Down Expand Up @@ -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, store.Request{}, action)
if err != nil {
err = stacktrace.Propagate(err, "Could not get subscription")
errResp := &restapi.ErrorResponse{Message: dsserr.Handle(ctx, err)}
Expand Down Expand Up @@ -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, store.Request{}, action)
if err != nil {

errResp := &restapi.ErrorResponse{Message: dsserr.Handle(ctx, err)}
Expand Down Expand Up @@ -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, store.Request{}, action)
if err != nil {
err = stacktrace.Propagate(err, "Could not delete subscription")
errResp := &restapi.ErrorResponse{Message: dsserr.Handle(ctx, err)}
Expand Down
5 changes: 3 additions & 2 deletions pkg/scd/uss_availability_handler.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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"
"github.com/interuss/dss/pkg/store"
"github.com/interuss/stacktrace"
"github.com/jackc/pgx/v5"
)
Expand Down Expand Up @@ -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, store.Request{}, action)
if err != nil {
// In case of older DB versions where availability table doesn't exist
if strings.Contains(err.Error(), "does not exist") {
Expand Down Expand Up @@ -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, store.Request{}, action)
if err != nil {
// In case of older DB versions where availability table doesn't exist
if strings.Contains(err.Error(), "does not exist") {
Expand Down
6 changes: 3 additions & 3 deletions pkg/sqlstore/store.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -176,13 +176,13 @@ 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, _ store.Request, f func(context.Context, R) error) (any, error) {
if s.Version.Type == Yugabyte {
return s.transactYugabyte(ctx, f)
return nil, s.transactYugabyte(ctx, f)
}

ctx = crdb.WithMaxRetries(ctx, s.maxRetries)
return crdbpgx.ExecuteTx(ctx, s.Pool, pgx.TxOptions{IsoLevel: pgx.Serializable}, func(tx pgx.Tx) error {
return nil, crdbpgx.ExecuteTx(ctx, s.Pool, pgx.TxOptions{IsoLevel: pgx.Serializable}, func(tx pgx.Tx) error {
return f(ctx, s.newRepo(tx))
})
}
Expand Down
14 changes: 11 additions & 3 deletions pkg/store/store.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -7,16 +7,24 @@ import (
"github.com/interuss/stacktrace"
)

// Request carries the information needed by the Raftstore to handle a request.
type Request struct {
RequestType string
Payload any
}

// store.Store is the generic means to access and interact with any type of data backing the DSS
// may ever use, by obtaining a means to perform R-specific (repo type) operations.
type Store[R any] interface {
io.Closer
// Obtain a Repo (repo type R) that doesn't need transactional guarantees (for instance,
// read-only).
Interact(context.Context) (R, error)
// Attempt to apply the operations in f to the R Repo it is supplied. All operations performed
// on the R Repo by f will be applied or rejected atomically.
Transact(ctx context.Context, f func(context.Context, R) error) error
// Transact atomically applies or rejects a request.
// f specifies the operations to be performed on the repo.
// request specifies the type and payload of request.
// any is the request result.
Transact(ctx context.Context, request Request, f func(context.Context, R) error) (any, error)
}

const (
Expand Down
Loading