From 2381693fc6566e4c78c71e0799d750a41cae5007 Mon Sep 17 00:00:00 2001 From: Dmitry Date: Tue, 7 Nov 2023 12:54:36 +0100 Subject: [PATCH 01/12] iterate over database --- go.mod | 2 +- go.sum | 2 + sql/atxs/atxs.go | 100 +++++++++++++++++++++++++++++++++++++++++++++++ 3 files changed, 103 insertions(+), 1 deletion(-) diff --git a/go.mod b/go.mod index 26e0e59d2c..f79f748ecb 100644 --- a/go.mod +++ b/go.mod @@ -33,7 +33,7 @@ require ( github.com/prometheus/common v0.45.0 github.com/santhosh-tekuri/jsonschema/v5 v5.3.1 github.com/seehuhn/mt19937 v1.0.0 - github.com/spacemeshos/api/release/go v1.24.0 + github.com/spacemeshos/api/release/go v1.24.1-0.20231107112546-a48bc0143bd3 github.com/spacemeshos/economics v0.1.1 github.com/spacemeshos/fixed v0.1.1 github.com/spacemeshos/go-scale v1.1.12 diff --git a/go.sum b/go.sum index b9c20e7d58..08e206d110 100644 --- a/go.sum +++ b/go.sum @@ -643,6 +643,8 @@ github.com/sourcegraph/conc v0.3.0/go.mod h1:Sdozi7LEKbFPqYX2/J+iBAM6HpqSLTASQIK github.com/sourcegraph/syntaxhighlight v0.0.0-20170531221838-bd320f5d308e/go.mod h1:HuIsMU8RRBOtsCgI77wP899iHVBQpCmg4ErYMZB+2IA= github.com/spacemeshos/api/release/go v1.24.0 h1:uB4ZdmbHpodtpTyb7tDGxkTyb0nqr6fOIKv47SUChwk= github.com/spacemeshos/api/release/go v1.24.0/go.mod h1:SwqQxbhAF7tN3Qr34eVzczCB3KyTbkHH12U82eqfy6M= +github.com/spacemeshos/api/release/go v1.24.1-0.20231107112546-a48bc0143bd3 h1:SUjpIqNkAUIHTjUEq53/PYPMZANiwPJiwdzihdxHORU= +github.com/spacemeshos/api/release/go v1.24.1-0.20231107112546-a48bc0143bd3/go.mod h1:SwqQxbhAF7tN3Qr34eVzczCB3KyTbkHH12U82eqfy6M= github.com/spacemeshos/economics v0.1.1 h1:BPgMoTaeQ05ME6wEA1+MvXMp+wvXr51bIuN23thrCAk= github.com/spacemeshos/economics v0.1.1/go.mod h1:76nTjugYRiQ5/eD/DQs2dXPPilp28URMswUKncfdanY= github.com/spacemeshos/fixed v0.1.1 h1:N1y4SUpq1EV+IdJrWJwUCt1oBFzeru/VKVcBsvPc2Fk= diff --git a/sql/atxs/atxs.go b/sql/atxs/atxs.go index fc024e98cd..bf1ab674dd 100644 --- a/sql/atxs/atxs.go +++ b/sql/atxs/atxs.go @@ -4,6 +4,7 @@ import ( "fmt" "time" + spacemeshv1 "github.com/spacemeshos/api/release/go/spacemesh/v1" "github.com/spacemeshos/go-spacemesh/codec" "github.com/spacemeshos/go-spacemesh/common/types" "github.com/spacemeshos/go-spacemesh/sql" @@ -469,3 +470,102 @@ func IterateAtxs(db sql.Executor, from, to types.EpochID, fn func(*types.Verifie } return derr } + +func IterateAtxGRPC( + db sql.Executor, + filter *spacemeshv1.ActivationStreamRequest, + fn func(*spacemeshv1.ActivationStreamResponse) bool, +) error { + query := queryFrom(filter) + bindings, err := bindingsFrom(filter) + if err != nil { + return err + } + var derr error + _, err = db.Exec(query, bindings, + decoder(func(atx *types.VerifiedActivationTx, err error) bool { + if atx != nil { + v1 := &spacemeshv1.ActivationV1{ + Id: atx.ID().Bytes(), + NodeId: atx.SmesherID.Bytes(), + Signature: atx.Signature.Bytes(), + PublishEpoch: atx.PublishEpoch.Uint32(), + Sequence: atx.Sequence, + PrevAtx: atx.PrevATXID[:], + PositioningAtx: atx.PositioningATX[:], + Coinbase: atx.Coinbase.String(), + Units: atx.NumUnits, + BaseTick: uint32(atx.BaseTickHeight()), + Ticks: uint32(atx.TickCount()), + } + return fn(&spacemeshv1.ActivationStreamResponse{V1: v1}) + } + derr = err + return derr == nil + })) + if err != nil { + return err + } + return derr +} + +// queryFrom and bindingsFrom should decode fields in the same order. + +func queryFrom(filter *spacemeshv1.ActivationStreamRequest) string { + query := fullQuery + if filter != nil { + return query + } + i := 1 + if filter.Epochs != nil { + query += fmt.Sprintf(" where epoch between ?%d and ?%d", i, i+1) + i += 2 + } + if filter.Id != nil { + query += fmt.Sprintf(" and id = ?%d", i) + i++ + } + if filter.NodeId != nil { + query += fmt.Sprintf(" and pubkey = ?%d", i) + i++ + } + if filter.Coinbase != nil { + query += fmt.Sprintf(" and coinbase = ?%d", i) + i++ + } + return query +} + +func bindingsFrom(filter *spacemeshv1.ActivationStreamRequest) (sql.Encoder, error) { + if filter == nil { + return nil, nil + } + var coinbase *types.Address + if filter.Coinbase != nil { + address, err := types.StringToAddress(filter.Coinbase.Coinbase) + if err != nil { + return nil, fmt.Errorf("invalid coinbase address: %w", err) + } + coinbase = &address + } + i := 1 + return func(stmt *sql.Statement) { + if filter.Epochs != nil { + stmt.BindInt64(i, int64(filter.Epochs.StartEpoch)) + stmt.BindInt64(i+1, int64(filter.Epochs.EndEpoch)) + i += 2 + } + if filter.Id != nil { + stmt.BindBytes(i, filter.Id.Id) + i++ + } + if filter.NodeId != nil { + stmt.BindBytes(i, filter.NodeId.NodeId) + i++ + } + if coinbase != nil { + stmt.BindBytes(i, coinbase.Bytes()) + i++ + } + }, nil +} From bab8ac6c755499dda4adf940537d2300ad004b43 Mon Sep 17 00:00:00 2001 From: Dmitry Date: Tue, 7 Nov 2023 13:38:06 +0100 Subject: [PATCH 02/12] add half baked streaming for atxs --- api/grpcserver/activation_service.go | 19 +++++++- api/grpcserver/activation_service_test.go | 14 +++--- go.mod | 2 +- go.sum | 2 + node/node.go | 1 + sql/atxs/atxs.go | 53 ++++++++++++++--------- sql/atxs/atxs_test.go | 11 +++++ 7 files changed, 72 insertions(+), 30 deletions(-) diff --git a/api/grpcserver/activation_service.go b/api/grpcserver/activation_service.go index 5591d2b6b4..19ec1418a9 100644 --- a/api/grpcserver/activation_service.go +++ b/api/grpcserver/activation_service.go @@ -17,15 +17,18 @@ import ( "github.com/spacemeshos/go-spacemesh/common/types" "github.com/spacemeshos/go-spacemesh/events" "github.com/spacemeshos/go-spacemesh/sql" + "github.com/spacemeshos/go-spacemesh/sql/atxs" ) type activationService struct { goldenAtx types.ATXID + db *sql.Database atxProvider atxProvider } -func NewActivationService(atxProvider atxProvider, goldenAtx types.ATXID) *activationService { +func NewActivationService(db *sql.Database, atxProvider atxProvider, goldenAtx types.ATXID) *activationService { return &activationService{ + db: db, goldenAtx: goldenAtx, atxProvider: atxProvider, } @@ -99,3 +102,17 @@ func (s *activationService) Highest(ctx context.Context, req *emptypb.Empty) (*p Atx: convertActivation(atx), }, nil } + +func (s *activationService) Stream(filter *pb.ActivationStreamRequest, stream pb.ActivationService_StreamServer) error { + if filter.Watch { + return status.Error(codes.InvalidArgument, "watch is not supported") + } + var ierr error + if err := atxs.IterateAtxGRPC(s.db, filter, func(atx *pb.ActivationStreamResponse) bool { + ierr = stream.Send(atx) + return ierr == nil + }); err != nil { + return status.Error(codes.Internal, err.Error()) + } + return nil +} diff --git a/api/grpcserver/activation_service_test.go b/api/grpcserver/activation_service_test.go index c6664ead29..bc0c25711e 100644 --- a/api/grpcserver/activation_service_test.go +++ b/api/grpcserver/activation_service_test.go @@ -23,7 +23,7 @@ func Test_Highest_ReturnsGoldenAtxOnError(t *testing.T) { ctrl := gomock.NewController(t) atxProvider := grpcserver.NewMockatxProvider(ctrl) goldenAtx := types.ATXID{2, 3, 4} - activationService := grpcserver.NewActivationService(atxProvider, goldenAtx) + activationService := grpcserver.NewActivationService(nil, atxProvider, goldenAtx) atxProvider.EXPECT().MaxHeightAtx().Return(types.EmptyATXID, errors.New("blah")) response, err := activationService.Highest(context.Background(), &emptypb.Empty{}) @@ -41,7 +41,7 @@ func Test_Highest_ReturnsMaxTickHeight(t *testing.T) { ctrl := gomock.NewController(t) atxProvider := grpcserver.NewMockatxProvider(ctrl) goldenAtx := types.ATXID{2, 3, 4} - activationService := grpcserver.NewActivationService(atxProvider, goldenAtx) + activationService := grpcserver.NewActivationService(nil, atxProvider, goldenAtx) atx := types.VerifiedActivationTx{ ActivationTx: &types.ActivationTx{ @@ -76,7 +76,7 @@ func Test_Highest_ReturnsMaxTickHeight(t *testing.T) { func TestGet_RejectInvalidAtxID(t *testing.T) { ctrl := gomock.NewController(t) atxProvider := grpcserver.NewMockatxProvider(ctrl) - activationService := grpcserver.NewActivationService(atxProvider, types.ATXID{1}) + activationService := grpcserver.NewActivationService(nil, atxProvider, types.ATXID{1}) _, err := activationService.Get(context.Background(), &pb.GetRequest{Id: []byte{1, 2, 3}}) require.Error(t, err) @@ -86,7 +86,7 @@ func TestGet_RejectInvalidAtxID(t *testing.T) { func TestGet_AtxNotPresent(t *testing.T) { ctrl := gomock.NewController(t) atxProvider := grpcserver.NewMockatxProvider(ctrl) - activationService := grpcserver.NewActivationService(atxProvider, types.ATXID{1}) + activationService := grpcserver.NewActivationService(nil, atxProvider, types.ATXID{1}) id := types.RandomATXID() atxProvider.EXPECT().GetFullAtx(id).Return(nil, nil) @@ -99,7 +99,7 @@ func TestGet_AtxNotPresent(t *testing.T) { func TestGet_AtxProviderReturnsFailure(t *testing.T) { ctrl := gomock.NewController(t) atxProvider := grpcserver.NewMockatxProvider(ctrl) - activationService := grpcserver.NewActivationService(atxProvider, types.ATXID{1}) + activationService := grpcserver.NewActivationService(nil, atxProvider, types.ATXID{1}) id := types.RandomATXID() atxProvider.EXPECT().GetFullAtx(id).Return(&types.VerifiedActivationTx{}, errors.New("")) @@ -112,7 +112,7 @@ func TestGet_AtxProviderReturnsFailure(t *testing.T) { func TestGet_HappyPath(t *testing.T) { ctrl := gomock.NewController(t) atxProvider := grpcserver.NewMockatxProvider(ctrl) - activationService := grpcserver.NewActivationService(atxProvider, types.ATXID{1}) + activationService := grpcserver.NewActivationService(nil, atxProvider, types.ATXID{1}) id := types.RandomATXID() atx := types.VerifiedActivationTx{ @@ -149,7 +149,7 @@ func TestGet_HappyPath(t *testing.T) { func TestGet_IdentityCanceled(t *testing.T) { ctrl := gomock.NewController(t) atxProvider := grpcserver.NewMockatxProvider(ctrl) - activationService := grpcserver.NewActivationService(atxProvider, types.ATXID{1}) + activationService := grpcserver.NewActivationService(nil, atxProvider, types.ATXID{1}) smesher, proof := grpcserver.BallotMalfeasance(t, sql.InMemory()) id := types.RandomATXID() diff --git a/go.mod b/go.mod index f79f748ecb..9b0453e929 100644 --- a/go.mod +++ b/go.mod @@ -33,7 +33,7 @@ require ( github.com/prometheus/common v0.45.0 github.com/santhosh-tekuri/jsonschema/v5 v5.3.1 github.com/seehuhn/mt19937 v1.0.0 - github.com/spacemeshos/api/release/go v1.24.1-0.20231107112546-a48bc0143bd3 + github.com/spacemeshos/api/release/go v1.24.1-0.20231107121856-9cea84c8888b github.com/spacemeshos/economics v0.1.1 github.com/spacemeshos/fixed v0.1.1 github.com/spacemeshos/go-scale v1.1.12 diff --git a/go.sum b/go.sum index 08e206d110..1fb3376e6e 100644 --- a/go.sum +++ b/go.sum @@ -645,6 +645,8 @@ github.com/spacemeshos/api/release/go v1.24.0 h1:uB4ZdmbHpodtpTyb7tDGxkTyb0nqr6f github.com/spacemeshos/api/release/go v1.24.0/go.mod h1:SwqQxbhAF7tN3Qr34eVzczCB3KyTbkHH12U82eqfy6M= github.com/spacemeshos/api/release/go v1.24.1-0.20231107112546-a48bc0143bd3 h1:SUjpIqNkAUIHTjUEq53/PYPMZANiwPJiwdzihdxHORU= github.com/spacemeshos/api/release/go v1.24.1-0.20231107112546-a48bc0143bd3/go.mod h1:SwqQxbhAF7tN3Qr34eVzczCB3KyTbkHH12U82eqfy6M= +github.com/spacemeshos/api/release/go v1.24.1-0.20231107121856-9cea84c8888b h1:zPRpcFS6YFhuZe8kXMsMAAKzS8/xWORlxZZ4VTirqyU= +github.com/spacemeshos/api/release/go v1.24.1-0.20231107121856-9cea84c8888b/go.mod h1:SwqQxbhAF7tN3Qr34eVzczCB3KyTbkHH12U82eqfy6M= github.com/spacemeshos/economics v0.1.1 h1:BPgMoTaeQ05ME6wEA1+MvXMp+wvXr51bIuN23thrCAk= github.com/spacemeshos/economics v0.1.1/go.mod h1:76nTjugYRiQ5/eD/DQs2dXPPilp28URMswUKncfdanY= github.com/spacemeshos/fixed v0.1.1 h1:N1y4SUpq1EV+IdJrWJwUCt1oBFzeru/VKVcBsvPc2Fk= diff --git a/node/node.go b/node/node.go index 35cea41b3e..14fcc10df9 100644 --- a/node/node.go +++ b/node/node.go @@ -1314,6 +1314,7 @@ func (app *App) initService( ), nil case grpcserver.Activation: return grpcserver.NewActivationService( + app.cachedDB.Database, app.cachedDB, types.ATXID(app.Config.Genesis.GoldenATX()), ), nil diff --git a/sql/atxs/atxs.go b/sql/atxs/atxs.go index bf1ab674dd..456fceea47 100644 --- a/sql/atxs/atxs.go +++ b/sql/atxs/atxs.go @@ -476,13 +476,18 @@ func IterateAtxGRPC( filter *spacemeshv1.ActivationStreamRequest, fn func(*spacemeshv1.ActivationStreamResponse) bool, ) error { + full := fullQuery query := queryFrom(filter) + if query != "" { + full += " where " + query + } + full += " order by epoch asc" bindings, err := bindingsFrom(filter) if err != nil { return err } var derr error - _, err = db.Exec(query, bindings, + _, err = db.Exec(full, bindings, decoder(func(atx *types.VerifiedActivationTx, err error) bool { if atx != nil { v1 := &spacemeshv1.ActivationV1{ @@ -512,26 +517,31 @@ func IterateAtxGRPC( // queryFrom and bindingsFrom should decode fields in the same order. func queryFrom(filter *spacemeshv1.ActivationStreamRequest) string { - query := fullQuery - if filter != nil { + query := "" + if filter == nil { return query } i := 1 - if filter.Epochs != nil { - query += fmt.Sprintf(" where epoch between ?%d and ?%d", i, i+1) + and := "" + if filter.StartEpoch != 0 || filter.EndEpoch != 0 { + query += fmt.Sprintf(" epoch between ?%d and ?%d", i, i+1) + and = "and" i += 2 } - if filter.Id != nil { - query += fmt.Sprintf(" and id = ?%d", i) + if len(filter.Id) != 0 { + query += fmt.Sprintf(" %s id = ?%d", and, i) i++ + and = "and" } - if filter.NodeId != nil { - query += fmt.Sprintf(" and pubkey = ?%d", i) + if len(filter.NodeId) != 0 { + query += fmt.Sprintf(" %s pubkey = ?%d", and, i) i++ + and = "and" } - if filter.Coinbase != nil { - query += fmt.Sprintf(" and coinbase = ?%d", i) + if len(filter.Coinbase) != 0 { + query += fmt.Sprintf(" %s coinbase = ?%d", and, i) i++ + and = "and" } return query } @@ -541,8 +551,8 @@ func bindingsFrom(filter *spacemeshv1.ActivationStreamRequest) (sql.Encoder, err return nil, nil } var coinbase *types.Address - if filter.Coinbase != nil { - address, err := types.StringToAddress(filter.Coinbase.Coinbase) + if len(filter.Coinbase) != 0 { + address, err := types.StringToAddress(filter.Coinbase) if err != nil { return nil, fmt.Errorf("invalid coinbase address: %w", err) } @@ -550,17 +560,18 @@ func bindingsFrom(filter *spacemeshv1.ActivationStreamRequest) (sql.Encoder, err } i := 1 return func(stmt *sql.Statement) { - if filter.Epochs != nil { - stmt.BindInt64(i, int64(filter.Epochs.StartEpoch)) - stmt.BindInt64(i+1, int64(filter.Epochs.EndEpoch)) - i += 2 + if filter.StartEpoch != 0 || filter.EndEpoch != 0 { + stmt.BindInt64(i, int64(filter.StartEpoch)) + i++ + stmt.BindInt64(i, int64(filter.EndEpoch)) + i++ } - if filter.Id != nil { - stmt.BindBytes(i, filter.Id.Id) + if len(filter.Id) != 0 { + stmt.BindBytes(i, filter.Id) i++ } - if filter.NodeId != nil { - stmt.BindBytes(i, filter.NodeId.NodeId) + if len(filter.NodeId) != 0 { + stmt.BindBytes(i, filter.NodeId) i++ } if coinbase != nil { diff --git a/sql/atxs/atxs_test.go b/sql/atxs/atxs_test.go index d4f08bc25d..7325008ad0 100644 --- a/sql/atxs/atxs_test.go +++ b/sql/atxs/atxs_test.go @@ -1,12 +1,14 @@ package atxs_test import ( + "fmt" "os" "testing" "time" "github.com/stretchr/testify/require" + spacemeshv1 "github.com/spacemeshos/api/release/go/spacemesh/v1" "github.com/spacemeshos/go-spacemesh/activation" "github.com/spacemeshos/go-spacemesh/codec" "github.com/spacemeshos/go-spacemesh/common/types" @@ -745,6 +747,15 @@ func TestLatest(t *testing.T) { latest, err := atxs.LatestEpoch(db) require.NoError(t, err) require.EqualValues(t, tc.expect, latest) + + require.NoError(t, atxs.IterateAtxGRPC( + db, + &spacemeshv1.ActivationStreamRequest{StartEpoch: 7, EndEpoch: 7}, + func(asr *spacemeshv1.ActivationStreamResponse) bool { + fmt.Println(asr) + return true + }, + )) }) } } From a617a050614d3a6ae40e88a34035753a30827b32 Mon Sep 17 00:00:00 2001 From: Dmitry Date: Thu, 9 Nov 2023 09:01:45 +0100 Subject: [PATCH 03/12] refactor atxs db api --- api/grpcserver/activation_service.go | 21 +++- go.mod | 2 +- go.sum | 2 + sql/atxs/atxs.go | 151 ++++++++++++--------------- sql/atxs/atxs_test.go | 11 -- 5 files changed, 87 insertions(+), 100 deletions(-) diff --git a/api/grpcserver/activation_service.go b/api/grpcserver/activation_service.go index 19ec1418a9..949ac86e79 100644 --- a/api/grpcserver/activation_service.go +++ b/api/grpcserver/activation_service.go @@ -108,11 +108,28 @@ func (s *activationService) Stream(filter *pb.ActivationStreamRequest, stream pb return status.Error(codes.InvalidArgument, "watch is not supported") } var ierr error - if err := atxs.IterateAtxGRPC(s.db, filter, func(atx *pb.ActivationStreamResponse) bool { - ierr = stream.Send(atx) + if err := atxs.IterateAtxsOps(s.db, toOperations(filter), func(atx *types.VerifiedActivationTx) bool { + v1 := &pb.ActivationV1{ + Id: atx.ID().Bytes(), + NodeId: atx.SmesherID.Bytes(), + Signature: atx.Signature.Bytes(), + PublishEpoch: atx.PublishEpoch.Uint32(), + Sequence: atx.Sequence, + PrevAtx: atx.PrevATXID[:], + PositioningAtx: atx.PositioningATX[:], + Coinbase: atx.Coinbase.String(), + Units: atx.NumUnits, + BaseTick: uint32(atx.BaseTickHeight()), + Ticks: uint32(atx.TickCount()), + } + ierr = stream.Send(&pb.ActivationStreamResponse{V1: v1}) return ierr == nil }); err != nil { return status.Error(codes.Internal, err.Error()) } return nil } + +func toOperations(filter *pb.ActivationStreamRequest) atxs.Operations { + return atxs.Operations{} +} diff --git a/go.mod b/go.mod index 9b0453e929..43234745f8 100644 --- a/go.mod +++ b/go.mod @@ -33,7 +33,7 @@ require ( github.com/prometheus/common v0.45.0 github.com/santhosh-tekuri/jsonschema/v5 v5.3.1 github.com/seehuhn/mt19937 v1.0.0 - github.com/spacemeshos/api/release/go v1.24.1-0.20231107121856-9cea84c8888b + github.com/spacemeshos/api/release/go v1.24.1-0.20231109072853-7ec08711115c github.com/spacemeshos/economics v0.1.1 github.com/spacemeshos/fixed v0.1.1 github.com/spacemeshos/go-scale v1.1.12 diff --git a/go.sum b/go.sum index 1fb3376e6e..052816f82f 100644 --- a/go.sum +++ b/go.sum @@ -647,6 +647,8 @@ github.com/spacemeshos/api/release/go v1.24.1-0.20231107112546-a48bc0143bd3 h1:S github.com/spacemeshos/api/release/go v1.24.1-0.20231107112546-a48bc0143bd3/go.mod h1:SwqQxbhAF7tN3Qr34eVzczCB3KyTbkHH12U82eqfy6M= github.com/spacemeshos/api/release/go v1.24.1-0.20231107121856-9cea84c8888b h1:zPRpcFS6YFhuZe8kXMsMAAKzS8/xWORlxZZ4VTirqyU= github.com/spacemeshos/api/release/go v1.24.1-0.20231107121856-9cea84c8888b/go.mod h1:SwqQxbhAF7tN3Qr34eVzczCB3KyTbkHH12U82eqfy6M= +github.com/spacemeshos/api/release/go v1.24.1-0.20231109072853-7ec08711115c h1:9uA5Yh9J0xeHNvhv4ky1tOat7qAk0hJLPC12FYsFQNU= +github.com/spacemeshos/api/release/go v1.24.1-0.20231109072853-7ec08711115c/go.mod h1:SwqQxbhAF7tN3Qr34eVzczCB3KyTbkHH12U82eqfy6M= github.com/spacemeshos/economics v0.1.1 h1:BPgMoTaeQ05ME6wEA1+MvXMp+wvXr51bIuN23thrCAk= github.com/spacemeshos/economics v0.1.1/go.mod h1:76nTjugYRiQ5/eD/DQs2dXPPilp28URMswUKncfdanY= github.com/spacemeshos/fixed v0.1.1 h1:N1y4SUpq1EV+IdJrWJwUCt1oBFzeru/VKVcBsvPc2Fk= diff --git a/sql/atxs/atxs.go b/sql/atxs/atxs.go index 456fceea47..acbc24ac68 100644 --- a/sql/atxs/atxs.go +++ b/sql/atxs/atxs.go @@ -2,9 +2,9 @@ package atxs import ( "fmt" + "strconv" "time" - spacemeshv1 "github.com/spacemeshos/api/release/go/spacemesh/v1" "github.com/spacemeshos/go-spacemesh/codec" "github.com/spacemeshos/go-spacemesh/common/types" "github.com/spacemeshos/go-spacemesh/sql" @@ -471,39 +471,53 @@ func IterateAtxs(db sql.Executor, from, to types.EpochID, fn func(*types.Verifie return derr } -func IterateAtxGRPC( +type token string + +const ( + Eq token = "=" + NotEq token = "!=" + Gt token = ">" + Gte token = ">=" + Lt token = "<" + Lte token = "<=" + And token = "and" + Where token = "where" +) + +type field string + +const ( + Epoch field = "epoch" + Smesher field = "pubkey" + Coinbase field = "coinbase" + Id field = "id" +) + +type Op struct { + Field field + Token token + // Value will be type casted to one the expected types. + // Operation will panic if it doesn't match any of expected. + Value any +} + +type Operations struct { + Filter []Op + Other []Op +} + +func IterateAtxsOps( db sql.Executor, - filter *spacemeshv1.ActivationStreamRequest, - fn func(*spacemeshv1.ActivationStreamResponse) bool, + operations Operations, + fn func(*types.VerifiedActivationTx) bool, ) error { - full := fullQuery - query := queryFrom(filter) - if query != "" { - full += " where " + query - } - full += " order by epoch asc" - bindings, err := bindingsFrom(filter) - if err != nil { - return err - } var derr error - _, err = db.Exec(full, bindings, + _, err := db.Exec( + fullQuery+filterFrom(operations.Filter)+" order by epoch asc, id", + bindingsFrom(operations.Filter), decoder(func(atx *types.VerifiedActivationTx, err error) bool { if atx != nil { - v1 := &spacemeshv1.ActivationV1{ - Id: atx.ID().Bytes(), - NodeId: atx.SmesherID.Bytes(), - Signature: atx.Signature.Bytes(), - PublishEpoch: atx.PublishEpoch.Uint32(), - Sequence: atx.Sequence, - PrevAtx: atx.PrevATXID[:], - PositioningAtx: atx.PositioningATX[:], - Coinbase: atx.Coinbase.String(), - Units: atx.NumUnits, - BaseTick: uint32(atx.BaseTickHeight()), - Ticks: uint32(atx.TickCount()), - } - return fn(&spacemeshv1.ActivationStreamResponse{V1: v1}) + return fn(atx) } derr = err return derr == nil @@ -514,69 +528,34 @@ func IterateAtxGRPC( return derr } -// queryFrom and bindingsFrom should decode fields in the same order. - -func queryFrom(filter *spacemeshv1.ActivationStreamRequest) string { - query := "" - if filter == nil { - return query - } - i := 1 - and := "" - if filter.StartEpoch != 0 || filter.EndEpoch != 0 { - query += fmt.Sprintf(" epoch between ?%d and ?%d", i, i+1) - and = "and" - i += 2 - } - if len(filter.Id) != 0 { - query += fmt.Sprintf(" %s id = ?%d", and, i) - i++ - and = "and" - } - if len(filter.NodeId) != 0 { - query += fmt.Sprintf(" %s pubkey = ?%d", and, i) - i++ - and = "and" +func filterFrom(filter []Op) string { + if len(filter) == 0 { + return "" } - if len(filter.Coinbase) != 0 { - query += fmt.Sprintf(" %s coinbase = ?%d", and, i) - i++ - and = "and" + query := "where " + for i, op := range filter { + if i != 0 { + query += " " + string(And) + " " + } + query += string(op.Field) + " " + string(op.Token) + " ?" + strconv.Itoa(i+1) } return query } -func bindingsFrom(filter *spacemeshv1.ActivationStreamRequest) (sql.Encoder, error) { - if filter == nil { - return nil, nil +func bindingsFrom(filter []Op) sql.Encoder { + if len(filter) == 0 { + return nil } - var coinbase *types.Address - if len(filter.Coinbase) != 0 { - address, err := types.StringToAddress(filter.Coinbase) - if err != nil { - return nil, fmt.Errorf("invalid coinbase address: %w", err) - } - coinbase = &address - } - i := 1 return func(stmt *sql.Statement) { - if filter.StartEpoch != 0 || filter.EndEpoch != 0 { - stmt.BindInt64(i, int64(filter.StartEpoch)) - i++ - stmt.BindInt64(i, int64(filter.EndEpoch)) - i++ - } - if len(filter.Id) != 0 { - stmt.BindBytes(i, filter.Id) - i++ - } - if len(filter.NodeId) != 0 { - stmt.BindBytes(i, filter.NodeId) - i++ - } - if coinbase != nil { - stmt.BindBytes(i, coinbase.Bytes()) - i++ + for i, op := range filter { + switch value := op.Value.(type) { + case int64: + stmt.BindInt64(i+1, value) + case []byte: + stmt.BindBytes(i+1, value) + default: + panic(fmt.Sprintf("unexpected type %T", value)) + } } - }, nil + } } diff --git a/sql/atxs/atxs_test.go b/sql/atxs/atxs_test.go index 7325008ad0..d4f08bc25d 100644 --- a/sql/atxs/atxs_test.go +++ b/sql/atxs/atxs_test.go @@ -1,14 +1,12 @@ package atxs_test import ( - "fmt" "os" "testing" "time" "github.com/stretchr/testify/require" - spacemeshv1 "github.com/spacemeshos/api/release/go/spacemesh/v1" "github.com/spacemeshos/go-spacemesh/activation" "github.com/spacemeshos/go-spacemesh/codec" "github.com/spacemeshos/go-spacemesh/common/types" @@ -747,15 +745,6 @@ func TestLatest(t *testing.T) { latest, err := atxs.LatestEpoch(db) require.NoError(t, err) require.EqualValues(t, tc.expect, latest) - - require.NoError(t, atxs.IterateAtxGRPC( - db, - &spacemeshv1.ActivationStreamRequest{StartEpoch: 7, EndEpoch: 7}, - func(asr *spacemeshv1.ActivationStreamResponse) bool { - fmt.Println(asr) - return true - }, - )) }) } } From 4e9610dadecb9e046662a66990df5a56cadcc10b Mon Sep 17 00:00:00 2001 From: Dmitry Date: Thu, 9 Nov 2023 09:51:15 +0100 Subject: [PATCH 04/12] save progress --- api/grpcserver/activation_service.go | 53 +++++++++++++++++++++++-- api/grpcserver/v2/activation.go | 58 ++++++++++++++++++++++++++++ 2 files changed, 108 insertions(+), 3 deletions(-) create mode 100644 api/grpcserver/v2/activation.go diff --git a/api/grpcserver/activation_service.go b/api/grpcserver/activation_service.go index 949ac86e79..5298113423 100644 --- a/api/grpcserver/activation_service.go +++ b/api/grpcserver/activation_service.go @@ -107,8 +107,12 @@ func (s *activationService) Stream(filter *pb.ActivationStreamRequest, stream pb if filter.Watch { return status.Error(codes.InvalidArgument, "watch is not supported") } + ops, err := toOperations(filter) + if err != nil { + return status.Error(codes.InvalidArgument, err.Error()) + } var ierr error - if err := atxs.IterateAtxsOps(s.db, toOperations(filter), func(atx *types.VerifiedActivationTx) bool { + if err := atxs.IterateAtxsOps(s.db, ops, func(atx *types.VerifiedActivationTx) bool { v1 := &pb.ActivationV1{ Id: atx.ID().Bytes(), NodeId: atx.SmesherID.Bytes(), @@ -130,6 +134,49 @@ func (s *activationService) Stream(filter *pb.ActivationStreamRequest, stream pb return nil } -func toOperations(filter *pb.ActivationStreamRequest) atxs.Operations { - return atxs.Operations{} +func toOperations(filter *pb.ActivationStreamRequest) (atxs.Operations, error) { + ops := atxs.Operations{} + if filter == nil { + return ops, nil + } + if filter.NodeId != nil { + ops.Filter = append(ops.Filter, atxs.Op{ + Field: atxs.Smesher, + Token: atxs.Eq, + Value: filter.NodeId, + }) + } + if filter.Id != nil { + ops.Filter = append(ops.Filter, atxs.Op{ + Field: atxs.Id, + Token: atxs.Eq, + Value: filter.Id, + }) + } + if len(filter.Coinbase) > 0 { + addr, err := types.StringToAddress(filter.Coinbase) + if err != nil { + return atxs.Operations{}, err + } + ops.Filter = append(ops.Filter, atxs.Op{ + Field: atxs.Coinbase, + Token: atxs.Eq, + Value: addr.Bytes(), + }) + } + if filter.StartEpoch != 0 { + ops.Filter = append(ops.Filter, atxs.Op{ + Field: atxs.Epoch, + Token: atxs.Gte, + Value: int64(filter.StartEpoch), + }) + } + if filter.EndEpoch != 0 { + ops.Filter = append(ops.Filter, atxs.Op{ + Field: atxs.Epoch, + Token: atxs.Lte, + Value: int64(filter.EndEpoch), + }) + } + return atxs.Operations{}, nil } diff --git a/api/grpcserver/v2/activation.go b/api/grpcserver/v2/activation.go new file mode 100644 index 0000000000..55b033b68e --- /dev/null +++ b/api/grpcserver/v2/activation.go @@ -0,0 +1,58 @@ +package v2 + +import ( + "context" + + "github.com/grpc-ecosystem/grpc-gateway/v2/runtime" + "google.golang.org/grpc" + + spacemeshv2 "github.com/spacemeshos/api/release/go/spacemesh/v2" + + "github.com/spacemeshos/go-spacemesh/api/grpcserver" + "github.com/spacemeshos/go-spacemesh/sql" +) + +func NewActivationStreamService(db *sql.Database) *ActivationStreamService { + return &ActivationStreamService{db: db} +} + +type ActivationStreamService struct { + db *sql.Database +} + +var _ grpcserver.ServiceAPI = (*ActivationStreamService)(nil) + +func (s *ActivationStreamService) RegisterService(server *grpc.Server) { + spacemeshv2.RegisterActivationStreamServiceServer(server, s) +} + +func (s *ActivationStreamService) RegisterHandlerService(mux *runtime.ServeMux) error { + return spacemeshv2.RegisterActivationStreamServiceHandlerServer(context.Background(), mux, s) +} + +func (s *ActivationStreamService) String() string { + return "ActivationStreamService" +} + +func NewActivationService(db *sql.Database) *ActivationService { + return &ActivationService{db: db} +} + +type ActivationService struct { + db *sql.Database +} + +var _ grpcserver.ServiceAPI = (*ActivationService)(nil) + +func (s *ActivationService) RegisterService(server *grpc.Server) { + spacemeshv2.RegisterActivationServiceServer(server, s) +} + +func (s *ActivationService) RegisterHandlerService(mux *runtime.ServeMux) error { + return spacemeshv2.RegisterActivationServiceHandlerServer(context.Background(), mux, s) +} + +// String returns the service name. +func (s *ActivationService) String() string { + return "ActivationService" +} From a767855e8818589765c7ed3e3646b5367680ada1 Mon Sep 17 00:00:00 2001 From: Dmitry Date: Thu, 9 Nov 2023 10:48:04 +0100 Subject: [PATCH 05/12] register v2 --- api/grpcserver/activation_service.go | 83 +-------- api/grpcserver/activation_service_test.go | 14 +- api/grpcserver/config.go | 4 +- api/grpcserver/v2/activation.go | 217 ++++++++++++++++++++++ go.mod | 2 +- go.sum | 2 + node/node.go | 6 +- 7 files changed, 235 insertions(+), 93 deletions(-) diff --git a/api/grpcserver/activation_service.go b/api/grpcserver/activation_service.go index 5298113423..5591d2b6b4 100644 --- a/api/grpcserver/activation_service.go +++ b/api/grpcserver/activation_service.go @@ -17,18 +17,15 @@ import ( "github.com/spacemeshos/go-spacemesh/common/types" "github.com/spacemeshos/go-spacemesh/events" "github.com/spacemeshos/go-spacemesh/sql" - "github.com/spacemeshos/go-spacemesh/sql/atxs" ) type activationService struct { goldenAtx types.ATXID - db *sql.Database atxProvider atxProvider } -func NewActivationService(db *sql.Database, atxProvider atxProvider, goldenAtx types.ATXID) *activationService { +func NewActivationService(atxProvider atxProvider, goldenAtx types.ATXID) *activationService { return &activationService{ - db: db, goldenAtx: goldenAtx, atxProvider: atxProvider, } @@ -102,81 +99,3 @@ func (s *activationService) Highest(ctx context.Context, req *emptypb.Empty) (*p Atx: convertActivation(atx), }, nil } - -func (s *activationService) Stream(filter *pb.ActivationStreamRequest, stream pb.ActivationService_StreamServer) error { - if filter.Watch { - return status.Error(codes.InvalidArgument, "watch is not supported") - } - ops, err := toOperations(filter) - if err != nil { - return status.Error(codes.InvalidArgument, err.Error()) - } - var ierr error - if err := atxs.IterateAtxsOps(s.db, ops, func(atx *types.VerifiedActivationTx) bool { - v1 := &pb.ActivationV1{ - Id: atx.ID().Bytes(), - NodeId: atx.SmesherID.Bytes(), - Signature: atx.Signature.Bytes(), - PublishEpoch: atx.PublishEpoch.Uint32(), - Sequence: atx.Sequence, - PrevAtx: atx.PrevATXID[:], - PositioningAtx: atx.PositioningATX[:], - Coinbase: atx.Coinbase.String(), - Units: atx.NumUnits, - BaseTick: uint32(atx.BaseTickHeight()), - Ticks: uint32(atx.TickCount()), - } - ierr = stream.Send(&pb.ActivationStreamResponse{V1: v1}) - return ierr == nil - }); err != nil { - return status.Error(codes.Internal, err.Error()) - } - return nil -} - -func toOperations(filter *pb.ActivationStreamRequest) (atxs.Operations, error) { - ops := atxs.Operations{} - if filter == nil { - return ops, nil - } - if filter.NodeId != nil { - ops.Filter = append(ops.Filter, atxs.Op{ - Field: atxs.Smesher, - Token: atxs.Eq, - Value: filter.NodeId, - }) - } - if filter.Id != nil { - ops.Filter = append(ops.Filter, atxs.Op{ - Field: atxs.Id, - Token: atxs.Eq, - Value: filter.Id, - }) - } - if len(filter.Coinbase) > 0 { - addr, err := types.StringToAddress(filter.Coinbase) - if err != nil { - return atxs.Operations{}, err - } - ops.Filter = append(ops.Filter, atxs.Op{ - Field: atxs.Coinbase, - Token: atxs.Eq, - Value: addr.Bytes(), - }) - } - if filter.StartEpoch != 0 { - ops.Filter = append(ops.Filter, atxs.Op{ - Field: atxs.Epoch, - Token: atxs.Gte, - Value: int64(filter.StartEpoch), - }) - } - if filter.EndEpoch != 0 { - ops.Filter = append(ops.Filter, atxs.Op{ - Field: atxs.Epoch, - Token: atxs.Lte, - Value: int64(filter.EndEpoch), - }) - } - return atxs.Operations{}, nil -} diff --git a/api/grpcserver/activation_service_test.go b/api/grpcserver/activation_service_test.go index bc0c25711e..c6664ead29 100644 --- a/api/grpcserver/activation_service_test.go +++ b/api/grpcserver/activation_service_test.go @@ -23,7 +23,7 @@ func Test_Highest_ReturnsGoldenAtxOnError(t *testing.T) { ctrl := gomock.NewController(t) atxProvider := grpcserver.NewMockatxProvider(ctrl) goldenAtx := types.ATXID{2, 3, 4} - activationService := grpcserver.NewActivationService(nil, atxProvider, goldenAtx) + activationService := grpcserver.NewActivationService(atxProvider, goldenAtx) atxProvider.EXPECT().MaxHeightAtx().Return(types.EmptyATXID, errors.New("blah")) response, err := activationService.Highest(context.Background(), &emptypb.Empty{}) @@ -41,7 +41,7 @@ func Test_Highest_ReturnsMaxTickHeight(t *testing.T) { ctrl := gomock.NewController(t) atxProvider := grpcserver.NewMockatxProvider(ctrl) goldenAtx := types.ATXID{2, 3, 4} - activationService := grpcserver.NewActivationService(nil, atxProvider, goldenAtx) + activationService := grpcserver.NewActivationService(atxProvider, goldenAtx) atx := types.VerifiedActivationTx{ ActivationTx: &types.ActivationTx{ @@ -76,7 +76,7 @@ func Test_Highest_ReturnsMaxTickHeight(t *testing.T) { func TestGet_RejectInvalidAtxID(t *testing.T) { ctrl := gomock.NewController(t) atxProvider := grpcserver.NewMockatxProvider(ctrl) - activationService := grpcserver.NewActivationService(nil, atxProvider, types.ATXID{1}) + activationService := grpcserver.NewActivationService(atxProvider, types.ATXID{1}) _, err := activationService.Get(context.Background(), &pb.GetRequest{Id: []byte{1, 2, 3}}) require.Error(t, err) @@ -86,7 +86,7 @@ func TestGet_RejectInvalidAtxID(t *testing.T) { func TestGet_AtxNotPresent(t *testing.T) { ctrl := gomock.NewController(t) atxProvider := grpcserver.NewMockatxProvider(ctrl) - activationService := grpcserver.NewActivationService(nil, atxProvider, types.ATXID{1}) + activationService := grpcserver.NewActivationService(atxProvider, types.ATXID{1}) id := types.RandomATXID() atxProvider.EXPECT().GetFullAtx(id).Return(nil, nil) @@ -99,7 +99,7 @@ func TestGet_AtxNotPresent(t *testing.T) { func TestGet_AtxProviderReturnsFailure(t *testing.T) { ctrl := gomock.NewController(t) atxProvider := grpcserver.NewMockatxProvider(ctrl) - activationService := grpcserver.NewActivationService(nil, atxProvider, types.ATXID{1}) + activationService := grpcserver.NewActivationService(atxProvider, types.ATXID{1}) id := types.RandomATXID() atxProvider.EXPECT().GetFullAtx(id).Return(&types.VerifiedActivationTx{}, errors.New("")) @@ -112,7 +112,7 @@ func TestGet_AtxProviderReturnsFailure(t *testing.T) { func TestGet_HappyPath(t *testing.T) { ctrl := gomock.NewController(t) atxProvider := grpcserver.NewMockatxProvider(ctrl) - activationService := grpcserver.NewActivationService(nil, atxProvider, types.ATXID{1}) + activationService := grpcserver.NewActivationService(atxProvider, types.ATXID{1}) id := types.RandomATXID() atx := types.VerifiedActivationTx{ @@ -149,7 +149,7 @@ func TestGet_HappyPath(t *testing.T) { func TestGet_IdentityCanceled(t *testing.T) { ctrl := gomock.NewController(t) atxProvider := grpcserver.NewMockatxProvider(ctrl) - activationService := grpcserver.NewActivationService(nil, atxProvider, types.ATXID{1}) + activationService := grpcserver.NewActivationService(atxProvider, types.ATXID{1}) smesher, proof := grpcserver.BallotMalfeasance(t, sql.InMemory()) id := types.RandomATXID() diff --git a/api/grpcserver/config.go b/api/grpcserver/config.go index c08d63beff..488e78f611 100644 --- a/api/grpcserver/config.go +++ b/api/grpcserver/config.go @@ -39,9 +39,9 @@ const ( // DefaultConfig defines the default configuration options for api. func DefaultConfig() Config { return Config{ - PublicServices: []Service{Debug, GlobalState, Mesh, Transaction, Node, Activation}, + PublicServices: []Service{Debug, GlobalState, Mesh, Transaction, Node, Activation, "activation_v2"}, PublicListener: "0.0.0.0:9092", - PrivateServices: []Service{Admin, Smesher, Post}, + PrivateServices: []Service{Admin, Smesher, Post, "activation_stream_v2"}, PrivateListener: "127.0.0.1:9093", TLSServices: []Service{}, TLSListener: "0.0.0.0:9094", diff --git a/api/grpcserver/v2/activation.go b/api/grpcserver/v2/activation.go index 55b033b68e..07faa63107 100644 --- a/api/grpcserver/v2/activation.go +++ b/api/grpcserver/v2/activation.go @@ -5,11 +5,20 @@ import ( "github.com/grpc-ecosystem/grpc-gateway/v2/runtime" "google.golang.org/grpc" + "google.golang.org/grpc/codes" + "google.golang.org/grpc/status" spacemeshv2 "github.com/spacemeshos/api/release/go/spacemesh/v2" "github.com/spacemeshos/go-spacemesh/api/grpcserver" + "github.com/spacemeshos/go-spacemesh/common/types" "github.com/spacemeshos/go-spacemesh/sql" + "github.com/spacemeshos/go-spacemesh/sql/atxs" +) + +const ( + Activation = "activation_v2" + ActivationStream = "activation_stream_v2" ) func NewActivationStreamService(db *sql.Database) *ActivationStreamService { @@ -34,6 +43,77 @@ func (s *ActivationStreamService) String() string { return "ActivationStreamService" } +func (s *ActivationStreamService) Stream( + request *spacemeshv2.ActivationStreamRequest, + stream spacemeshv2.ActivationStreamService_StreamServer, +) error { + if request.Watch { + return status.Error(codes.InvalidArgument, "watch is not supported") + } + ops, err := toOperations(request) + if err != nil { + return status.Error(codes.InvalidArgument, err.Error()) + } + var ierr error + if err := atxs.IterateAtxsOps(s.db, ops, func(atx *types.VerifiedActivationTx) bool { + ierr = stream.Send(&spacemeshv2.Activation{Versioned: &spacemeshv2.Activation_V1{V1: toAtx(atx)}}) + return ierr == nil + }); err != nil { + return status.Error(codes.Internal, err.Error()) + } + return nil +} + +func (s *ActivationStreamService) StreamHeaders( + request *spacemeshv2.ActivationStreamRequest, + stream spacemeshv2.ActivationStreamService_StreamHeadersServer, +) error { + if request.Watch { + return status.Error(codes.InvalidArgument, "watch is not supported") + } + ops, err := toOperations(request) + if err != nil { + return status.Error(codes.InvalidArgument, err.Error()) + } + var ierr error + if err := atxs.IterateAtxsOps(s.db, ops, func(atx *types.VerifiedActivationTx) bool { + ierr = stream.Send(&spacemeshv2.ActivationHeader{Versioned: &spacemeshv2.ActivationHeader_V1{ + V1: toHeader(atx)}}) + return ierr == nil + }); err != nil { + return status.Error(codes.Internal, err.Error()) + } + return nil +} + +func toAtx(atx *types.VerifiedActivationTx) *spacemeshv2.ActivationV1 { + return &spacemeshv2.ActivationV1{ + Id: atx.ID().Bytes(), + NodeId: atx.SmesherID.Bytes(), + Signature: atx.Signature.Bytes(), + PublishEpoch: atx.PublishEpoch.Uint32(), + Sequence: atx.Sequence, + PrevAtx: atx.PrevATXID[:], + PositioningAtx: atx.PositioningATX[:], + Coinbase: atx.Coinbase.String(), + Units: atx.NumUnits, + BaseHeight: uint32(atx.BaseTickHeight()), + Ticks: uint32(atx.TickCount()), + } +} + +func toHeader(atx *types.VerifiedActivationTx) *spacemeshv2.ActivationHeaderV1 { + return &spacemeshv2.ActivationHeaderV1{ + Id: atx.ID().Bytes(), + NodeId: atx.SmesherID.Bytes(), + PublishEpoch: atx.PublishEpoch.Uint32(), + Coinbase: atx.Coinbase.String(), + Units: atx.NumUnits, + BaseHeight: uint32(atx.BaseTickHeight()), + Ticks: uint32(atx.TickCount()), + } +} + func NewActivationService(db *sql.Database) *ActivationService { return &ActivationService{db: db} } @@ -56,3 +136,140 @@ func (s *ActivationService) RegisterHandlerService(mux *runtime.ServeMux) error func (s *ActivationService) String() string { return "ActivationService" } + +func (s *ActivationService) List( + ctx context.Context, + request *spacemeshv2.ActivationRequest, +) (*spacemeshv2.ActivationList, error) { + ops, err := toOperations2(request) + if err != nil { + return nil, status.Error(codes.InvalidArgument, err.Error()) + } + // every full atx is ~1KB. 100 atxs is ~100KB. + if request.Limit > 100 { + return nil, status.Error(codes.InvalidArgument, "limit is capped at 100") + } + rst := make([]*spacemeshv2.Activation, 0, request.Limit) + if err := atxs.IterateAtxsOps(s.db, ops, func(atx *types.VerifiedActivationTx) bool { + rst = append(rst, &spacemeshv2.Activation{Versioned: &spacemeshv2.Activation_V1{V1: toAtx(atx)}}) + return true + }); err != nil { + return nil, status.Error(codes.Internal, err.Error()) + } + return &spacemeshv2.ActivationList{Activations: rst}, nil +} + +func (s *ActivationService) ListHeaders( + ctx context.Context, + request *spacemeshv2.ActivationRequest, +) (*spacemeshv2.ActivationHeaderList, error) { + ops, err := toOperations2(request) + if err != nil { + return nil, status.Error(codes.InvalidArgument, err.Error()) + } + if request.Limit > 10000 { + return nil, status.Error(codes.InvalidArgument, "limit is capped at 10000") + } + rst := make([]*spacemeshv2.ActivationHeader, 0, request.Limit) + if err := atxs.IterateAtxsOps(s.db, ops, func(atx *types.VerifiedActivationTx) bool { + rst = append(rst, &spacemeshv2.ActivationHeader{Versioned: &spacemeshv2.ActivationHeader_V1{V1: toHeader(atx)}}) + return true + }); err != nil { + return nil, status.Error(codes.Internal, err.Error()) + } + return &spacemeshv2.ActivationHeaderList{Headers: rst}, nil +} + +func toOperations(filter *spacemeshv2.ActivationStreamRequest) (atxs.Operations, error) { + ops := atxs.Operations{} + if filter == nil { + return ops, nil + } + if filter.NodeId != nil { + ops.Filter = append(ops.Filter, atxs.Op{ + Field: atxs.Smesher, + Token: atxs.Eq, + Value: filter.NodeId, + }) + } + if filter.Id != nil { + ops.Filter = append(ops.Filter, atxs.Op{ + Field: atxs.Id, + Token: atxs.Eq, + Value: filter.Id, + }) + } + if len(filter.Coinbase) > 0 { + addr, err := types.StringToAddress(filter.Coinbase) + if err != nil { + return atxs.Operations{}, err + } + ops.Filter = append(ops.Filter, atxs.Op{ + Field: atxs.Coinbase, + Token: atxs.Eq, + Value: addr.Bytes(), + }) + } + if filter.StartEpoch != 0 { + ops.Filter = append(ops.Filter, atxs.Op{ + Field: atxs.Epoch, + Token: atxs.Gte, + Value: int64(filter.StartEpoch), + }) + } + if filter.EndEpoch != 0 { + ops.Filter = append(ops.Filter, atxs.Op{ + Field: atxs.Epoch, + Token: atxs.Lte, + Value: int64(filter.EndEpoch), + }) + } + return atxs.Operations{}, nil +} + +func toOperations2(filter *spacemeshv2.ActivationRequest) (atxs.Operations, error) { + ops := atxs.Operations{} + if filter == nil { + return ops, nil + } + if filter.NodeId != nil { + ops.Filter = append(ops.Filter, atxs.Op{ + Field: atxs.Smesher, + Token: atxs.Eq, + Value: filter.NodeId, + }) + } + if filter.Id != nil { + ops.Filter = append(ops.Filter, atxs.Op{ + Field: atxs.Id, + Token: atxs.Eq, + Value: filter.Id, + }) + } + if len(filter.Coinbase) > 0 { + addr, err := types.StringToAddress(filter.Coinbase) + if err != nil { + return atxs.Operations{}, err + } + ops.Filter = append(ops.Filter, atxs.Op{ + Field: atxs.Coinbase, + Token: atxs.Eq, + Value: addr.Bytes(), + }) + } + if filter.StartEpoch != 0 { + ops.Filter = append(ops.Filter, atxs.Op{ + Field: atxs.Epoch, + Token: atxs.Gte, + Value: int64(filter.StartEpoch), + }) + } + if filter.EndEpoch != 0 { + ops.Filter = append(ops.Filter, atxs.Op{ + Field: atxs.Epoch, + Token: atxs.Lte, + Value: int64(filter.EndEpoch), + }) + } + return atxs.Operations{}, nil +} diff --git a/go.mod b/go.mod index 43234745f8..366bc6360d 100644 --- a/go.mod +++ b/go.mod @@ -33,7 +33,7 @@ require ( github.com/prometheus/common v0.45.0 github.com/santhosh-tekuri/jsonschema/v5 v5.3.1 github.com/seehuhn/mt19937 v1.0.0 - github.com/spacemeshos/api/release/go v1.24.1-0.20231109072853-7ec08711115c + github.com/spacemeshos/api/release/go v1.24.1-0.20231109094211-d9b5d4ad4b20 github.com/spacemeshos/economics v0.1.1 github.com/spacemeshos/fixed v0.1.1 github.com/spacemeshos/go-scale v1.1.12 diff --git a/go.sum b/go.sum index 052816f82f..4bf7d156e4 100644 --- a/go.sum +++ b/go.sum @@ -649,6 +649,8 @@ github.com/spacemeshos/api/release/go v1.24.1-0.20231107121856-9cea84c8888b h1:z github.com/spacemeshos/api/release/go v1.24.1-0.20231107121856-9cea84c8888b/go.mod h1:SwqQxbhAF7tN3Qr34eVzczCB3KyTbkHH12U82eqfy6M= github.com/spacemeshos/api/release/go v1.24.1-0.20231109072853-7ec08711115c h1:9uA5Yh9J0xeHNvhv4ky1tOat7qAk0hJLPC12FYsFQNU= github.com/spacemeshos/api/release/go v1.24.1-0.20231109072853-7ec08711115c/go.mod h1:SwqQxbhAF7tN3Qr34eVzczCB3KyTbkHH12U82eqfy6M= +github.com/spacemeshos/api/release/go v1.24.1-0.20231109094211-d9b5d4ad4b20 h1:Ih/n+v1cprkGbcdKzQ4kyoSBE8jGjNSo6rZk1ifZHtI= +github.com/spacemeshos/api/release/go v1.24.1-0.20231109094211-d9b5d4ad4b20/go.mod h1:SwqQxbhAF7tN3Qr34eVzczCB3KyTbkHH12U82eqfy6M= github.com/spacemeshos/economics v0.1.1 h1:BPgMoTaeQ05ME6wEA1+MvXMp+wvXr51bIuN23thrCAk= github.com/spacemeshos/economics v0.1.1/go.mod h1:76nTjugYRiQ5/eD/DQs2dXPPilp28URMswUKncfdanY= github.com/spacemeshos/fixed v0.1.1 h1:N1y4SUpq1EV+IdJrWJwUCt1oBFzeru/VKVcBsvPc2Fk= diff --git a/node/node.go b/node/node.go index 14fcc10df9..e665de6d03 100644 --- a/node/node.go +++ b/node/node.go @@ -31,6 +31,7 @@ import ( "github.com/spacemeshos/go-spacemesh/activation" "github.com/spacemeshos/go-spacemesh/api/grpcserver" + v2 "github.com/spacemeshos/go-spacemesh/api/grpcserver/v2" "github.com/spacemeshos/go-spacemesh/atxsdata" "github.com/spacemeshos/go-spacemesh/beacon" "github.com/spacemeshos/go-spacemesh/blocks" @@ -1314,10 +1315,13 @@ func (app *App) initService( ), nil case grpcserver.Activation: return grpcserver.NewActivationService( - app.cachedDB.Database, app.cachedDB, types.ATXID(app.Config.Genesis.GoldenATX()), ), nil + case v2.Activation: + return v2.NewActivationService(app.db), nil + case v2.ActivationStream: + return v2.NewActivationStreamService(app.db), nil } return nil, fmt.Errorf("unknown service %s", svc) } From d30636f7c3b214e7e585b973680f31ba05d612dc Mon Sep 17 00:00:00 2001 From: Dmitry Date: Thu, 9 Nov 2023 12:03:23 +0100 Subject: [PATCH 06/12] support for offset and limit --- api/grpcserver/v2/activation.go | 77 ++++++++++++--------------------- sql/atxs/atxs.go | 26 ++++++----- 2 files changed, 42 insertions(+), 61 deletions(-) diff --git a/api/grpcserver/v2/activation.go b/api/grpcserver/v2/activation.go index 07faa63107..6d9fc35846 100644 --- a/api/grpcserver/v2/activation.go +++ b/api/grpcserver/v2/activation.go @@ -50,7 +50,7 @@ func (s *ActivationStreamService) Stream( if request.Watch { return status.Error(codes.InvalidArgument, "watch is not supported") } - ops, err := toOperations(request) + ops, err := toOperations(toRequest(request)) if err != nil { return status.Error(codes.InvalidArgument, err.Error()) } @@ -71,7 +71,7 @@ func (s *ActivationStreamService) StreamHeaders( if request.Watch { return status.Error(codes.InvalidArgument, "watch is not supported") } - ops, err := toOperations(request) + ops, err := toOperations(toRequest(request)) if err != nil { return status.Error(codes.InvalidArgument, err.Error()) } @@ -141,13 +141,15 @@ func (s *ActivationService) List( ctx context.Context, request *spacemeshv2.ActivationRequest, ) (*spacemeshv2.ActivationList, error) { - ops, err := toOperations2(request) + ops, err := toOperations(request) if err != nil { return nil, status.Error(codes.InvalidArgument, err.Error()) } // every full atx is ~1KB. 100 atxs is ~100KB. if request.Limit > 100 { return nil, status.Error(codes.InvalidArgument, "limit is capped at 100") + } else if request.Limit == 0 { + return nil, status.Error(codes.InvalidArgument, "limit must be set to a value below 100") } rst := make([]*spacemeshv2.Activation, 0, request.Limit) if err := atxs.IterateAtxsOps(s.db, ops, func(atx *types.VerifiedActivationTx) bool { @@ -163,12 +165,14 @@ func (s *ActivationService) ListHeaders( ctx context.Context, request *spacemeshv2.ActivationRequest, ) (*spacemeshv2.ActivationHeaderList, error) { - ops, err := toOperations2(request) + ops, err := toOperations(request) if err != nil { return nil, status.Error(codes.InvalidArgument, err.Error()) } if request.Limit > 10000 { return nil, status.Error(codes.InvalidArgument, "limit is capped at 10000") + } else if request.Limit == 0 { + return nil, status.Error(codes.InvalidArgument, "limit must be set to a value below 10000") } rst := make([]*spacemeshv2.ActivationHeader, 0, request.Limit) if err := atxs.IterateAtxsOps(s.db, ops, func(atx *types.VerifiedActivationTx) bool { @@ -180,7 +184,17 @@ func (s *ActivationService) ListHeaders( return &spacemeshv2.ActivationHeaderList{Headers: rst}, nil } -func toOperations(filter *spacemeshv2.ActivationStreamRequest) (atxs.Operations, error) { +func toRequest(filter *spacemeshv2.ActivationStreamRequest) *spacemeshv2.ActivationRequest { + return &spacemeshv2.ActivationRequest{ + NodeId: filter.NodeId, + Id: filter.Id, + Coinbase: filter.Coinbase, + StartEpoch: filter.StartEpoch, + EndEpoch: filter.EndEpoch, + } +} + +func toOperations(filter *spacemeshv2.ActivationRequest) (atxs.Operations, error) { ops := atxs.Operations{} if filter == nil { return ops, nil @@ -224,52 +238,17 @@ func toOperations(filter *spacemeshv2.ActivationStreamRequest) (atxs.Operations, Value: int64(filter.EndEpoch), }) } - return atxs.Operations{}, nil -} - -func toOperations2(filter *spacemeshv2.ActivationRequest) (atxs.Operations, error) { - ops := atxs.Operations{} - if filter == nil { - return ops, nil - } - if filter.NodeId != nil { - ops.Filter = append(ops.Filter, atxs.Op{ - Field: atxs.Smesher, - Token: atxs.Eq, - Value: filter.NodeId, - }) - } - if filter.Id != nil { - ops.Filter = append(ops.Filter, atxs.Op{ - Field: atxs.Id, - Token: atxs.Eq, - Value: filter.Id, - }) - } - if len(filter.Coinbase) > 0 { - addr, err := types.StringToAddress(filter.Coinbase) - if err != nil { - return atxs.Operations{}, err - } - ops.Filter = append(ops.Filter, atxs.Op{ - Field: atxs.Coinbase, - Token: atxs.Eq, - Value: addr.Bytes(), - }) - } - if filter.StartEpoch != 0 { - ops.Filter = append(ops.Filter, atxs.Op{ - Field: atxs.Epoch, - Token: atxs.Gte, - Value: int64(filter.StartEpoch), + if filter.Offset != 0 { + ops.Other = append(ops.Other, atxs.Op{ + Field: atxs.Offset, + Value: int64(filter.Offset), }) } - if filter.EndEpoch != 0 { - ops.Filter = append(ops.Filter, atxs.Op{ - Field: atxs.Epoch, - Token: atxs.Lte, - Value: int64(filter.EndEpoch), + if filter.Limit != 0 { + ops.Other = append(ops.Other, atxs.Op{ + Field: atxs.Limit, + Value: int64(filter.Limit), }) } - return atxs.Operations{}, nil + return ops, nil } diff --git a/sql/atxs/atxs.go b/sql/atxs/atxs.go index acbc24ac68..b19a1b43b7 100644 --- a/sql/atxs/atxs.go +++ b/sql/atxs/atxs.go @@ -491,6 +491,8 @@ const ( Smesher field = "pubkey" Coinbase field = "coinbase" Id field = "id" + Offset field = "offset" + Limit field = "limit" ) type Op struct { @@ -513,8 +515,8 @@ func IterateAtxsOps( ) error { var derr error _, err := db.Exec( - fullQuery+filterFrom(operations.Filter)+" order by epoch asc, id", - bindingsFrom(operations.Filter), + fullQuery+filterFrom(operations)+" order by epoch asc, id", + bindingsFrom(operations), decoder(func(atx *types.VerifiedActivationTx, err error) bool { if atx != nil { return fn(atx) @@ -528,26 +530,26 @@ func IterateAtxsOps( return derr } -func filterFrom(filter []Op) string { - if len(filter) == 0 { - return "" +func filterFrom(operations Operations) string { + query := " " + if len(operations.Filter) > 0 { + query = "where " } - query := "where " - for i, op := range filter { + for i, op := range operations.Filter { if i != 0 { query += " " + string(And) + " " } query += string(op.Field) + " " + string(op.Token) + " ?" + strconv.Itoa(i+1) } + for i, op := range operations.Other { + query += " " + string(op.Field) + " " + string(op.Token) + " ?" + strconv.Itoa(i+1+len(operations.Filter)) + } return query } -func bindingsFrom(filter []Op) sql.Encoder { - if len(filter) == 0 { - return nil - } +func bindingsFrom(operations Operations) sql.Encoder { return func(stmt *sql.Statement) { - for i, op := range filter { + for i, op := range append(operations.Filter, operations.Other...) { switch value := op.Value.(type) { case int64: stmt.BindInt64(i+1, value) From 7ff7f6754b263fe47e74ad10a065bb18d2636f54 Mon Sep 17 00:00:00 2001 From: Dmitry Date: Thu, 9 Nov 2023 12:13:37 +0100 Subject: [PATCH 07/12] encode the rest of the fields --- api/grpcserver/v2/activation.go | 40 ++++++++++++++++++++++++++++++++- go.mod | 2 +- go.sum | 2 ++ 3 files changed, 42 insertions(+), 2 deletions(-) diff --git a/api/grpcserver/v2/activation.go b/api/grpcserver/v2/activation.go index 6d9fc35846..34462ffb21 100644 --- a/api/grpcserver/v2/activation.go +++ b/api/grpcserver/v2/activation.go @@ -87,7 +87,7 @@ func (s *ActivationStreamService) StreamHeaders( } func toAtx(atx *types.VerifiedActivationTx) *spacemeshv2.ActivationV1 { - return &spacemeshv2.ActivationV1{ + v1 := &spacemeshv2.ActivationV1{ Id: atx.ID().Bytes(), NodeId: atx.SmesherID.Bytes(), Signature: atx.Signature.Bytes(), @@ -100,6 +100,44 @@ func toAtx(atx *types.VerifiedActivationTx) *spacemeshv2.ActivationV1 { BaseHeight: uint32(atx.BaseTickHeight()), Ticks: uint32(atx.TickCount()), } + if atx.CommitmentATX != nil { + v1.CommittmentAtx = atx.CommitmentATX.Bytes() + } + if atx.VRFNonce != nil { + v1.VrfPostIndex = &spacemeshv2.VRFPostIndex{ + Nonce: uint64(*atx.VRFNonce), + } + } + if atx.InitialPost != nil { + v1.InitialPost = &spacemeshv2.Post{ + Nonce: atx.InitialPost.Nonce, + Indices: atx.InitialPost.Indices, + Pow: atx.InitialPost.Pow, + } + } + if nipost := atx.NIPost; nipost != nil { + if nipost.Post != nil { + v1.Post = &spacemeshv2.Post{ + Nonce: nipost.Post.Nonce, + Indices: nipost.Post.Indices, + Pow: nipost.Post.Pow, + } + } + if nipost.PostMetadata != nil { + v1.PostMeta = &spacemeshv2.PostMeta{ + Challenge: nipost.PostMetadata.Challenge, + Labels: nipost.PostMetadata.LabelsPerUnit, + } + } + v1.PoetProof = &spacemeshv2.PoetProof{ + ProofNodes: make([][]byte, len(nipost.Membership.Nodes)), + Leaf: nipost.Membership.LeafIndex, + } + for i, node := range nipost.Membership.Nodes { + v1.PoetProof.ProofNodes[i] = node.Bytes() + } + } + return v1 } func toHeader(atx *types.VerifiedActivationTx) *spacemeshv2.ActivationHeaderV1 { diff --git a/go.mod b/go.mod index 366bc6360d..6d3ec4b89e 100644 --- a/go.mod +++ b/go.mod @@ -33,7 +33,7 @@ require ( github.com/prometheus/common v0.45.0 github.com/santhosh-tekuri/jsonschema/v5 v5.3.1 github.com/seehuhn/mt19937 v1.0.0 - github.com/spacemeshos/api/release/go v1.24.1-0.20231109094211-d9b5d4ad4b20 + github.com/spacemeshos/api/release/go v1.24.1-0.20231109110859-b10d491252b4 github.com/spacemeshos/economics v0.1.1 github.com/spacemeshos/fixed v0.1.1 github.com/spacemeshos/go-scale v1.1.12 diff --git a/go.sum b/go.sum index 4bf7d156e4..30500bb035 100644 --- a/go.sum +++ b/go.sum @@ -651,6 +651,8 @@ github.com/spacemeshos/api/release/go v1.24.1-0.20231109072853-7ec08711115c h1:9 github.com/spacemeshos/api/release/go v1.24.1-0.20231109072853-7ec08711115c/go.mod h1:SwqQxbhAF7tN3Qr34eVzczCB3KyTbkHH12U82eqfy6M= github.com/spacemeshos/api/release/go v1.24.1-0.20231109094211-d9b5d4ad4b20 h1:Ih/n+v1cprkGbcdKzQ4kyoSBE8jGjNSo6rZk1ifZHtI= github.com/spacemeshos/api/release/go v1.24.1-0.20231109094211-d9b5d4ad4b20/go.mod h1:SwqQxbhAF7tN3Qr34eVzczCB3KyTbkHH12U82eqfy6M= +github.com/spacemeshos/api/release/go v1.24.1-0.20231109110859-b10d491252b4 h1:8AUxvXZSFWYYSMbRp0zuPUUygcQMDDWXgmj1Uc9VtP8= +github.com/spacemeshos/api/release/go v1.24.1-0.20231109110859-b10d491252b4/go.mod h1:SwqQxbhAF7tN3Qr34eVzczCB3KyTbkHH12U82eqfy6M= github.com/spacemeshos/economics v0.1.1 h1:BPgMoTaeQ05ME6wEA1+MvXMp+wvXr51bIuN23thrCAk= github.com/spacemeshos/economics v0.1.1/go.mod h1:76nTjugYRiQ5/eD/DQs2dXPPilp28URMswUKncfdanY= github.com/spacemeshos/fixed v0.1.1 h1:N1y4SUpq1EV+IdJrWJwUCt1oBFzeru/VKVcBsvPc2Fk= From f4abc11d2451d2c9806c9de4fe47644983d14122 Mon Sep 17 00:00:00 2001 From: Dmitry Date: Thu, 9 Nov 2023 12:26:16 +0100 Subject: [PATCH 08/12] allow to watch --- api/grpcserver/v2/activation.go | 73 +++++++++++++++++++++++++++++++-- 1 file changed, 69 insertions(+), 4 deletions(-) diff --git a/api/grpcserver/v2/activation.go b/api/grpcserver/v2/activation.go index 34462ffb21..44172c5a07 100644 --- a/api/grpcserver/v2/activation.go +++ b/api/grpcserver/v2/activation.go @@ -2,16 +2,20 @@ package v2 import ( "context" + "errors" + "io" "github.com/grpc-ecosystem/grpc-gateway/v2/runtime" "google.golang.org/grpc" "google.golang.org/grpc/codes" + "google.golang.org/grpc/metadata" "google.golang.org/grpc/status" spacemeshv2 "github.com/spacemeshos/api/release/go/spacemesh/v2" "github.com/spacemeshos/go-spacemesh/api/grpcserver" "github.com/spacemeshos/go-spacemesh/common/types" + "github.com/spacemeshos/go-spacemesh/events" "github.com/spacemeshos/go-spacemesh/sql" "github.com/spacemeshos/go-spacemesh/sql/atxs" ) @@ -47,8 +51,18 @@ func (s *ActivationStreamService) Stream( request *spacemeshv2.ActivationStreamRequest, stream spacemeshv2.ActivationStreamService_StreamServer, ) error { + // TODO(dshulyak) implement matcher based on filter + var sub *events.BufferedSubscription[events.ActivationTx] if request.Watch { - return status.Error(codes.InvalidArgument, "watch is not supported") + var err error + sub, err = events.Subscribe[events.ActivationTx]() + if err != nil { + return status.Error(codes.Internal, err.Error()) + } + defer sub.Close() + if err := stream.SendHeader(metadata.MD{}); err != nil { + return status.Errorf(codes.Unavailable, "can't send header") + } } ops, err := toOperations(toRequest(request)) if err != nil { @@ -61,15 +75,47 @@ func (s *ActivationStreamService) Stream( }); err != nil { return status.Error(codes.Internal, err.Error()) } - return nil + if sub == nil { + return nil + } + for { + select { + case <-stream.Context().Done(): + return nil + case <-sub.Full(): + return status.Error(codes.Canceled, "buffer overflow") + case rst := <-sub.Out(): + if err := stream.Send(&spacemeshv2.Activation{ + Versioned: &spacemeshv2.Activation_V1{V1: toAtx(rst.VerifiedActivationTx)}}, + ); err != nil { + if errors.Is(err, io.EOF) { + return nil + } + return status.Error(codes.Internal, err.Error()) + } + } + } } func (s *ActivationStreamService) StreamHeaders( request *spacemeshv2.ActivationStreamRequest, stream spacemeshv2.ActivationStreamService_StreamHeadersServer, ) error { + // TODO(dshulyak) the code below is almost the same as code in Stream + // it can be refactored by implementing generic with toAtx/toHeader + + // TODO(dshulyak) implement matcher based on filter + var sub *events.BufferedSubscription[events.ActivationTx] if request.Watch { - return status.Error(codes.InvalidArgument, "watch is not supported") + var err error + sub, err = events.Subscribe[events.ActivationTx]() + if err != nil { + return status.Error(codes.Internal, err.Error()) + } + defer sub.Close() + if err := stream.SendHeader(metadata.MD{}); err != nil { + return status.Errorf(codes.Unavailable, "can't send header") + } } ops, err := toOperations(toRequest(request)) if err != nil { @@ -83,7 +129,26 @@ func (s *ActivationStreamService) StreamHeaders( }); err != nil { return status.Error(codes.Internal, err.Error()) } - return nil + if sub == nil { + return nil + } + for { + select { + case <-stream.Context().Done(): + return nil + case <-sub.Full(): + return status.Error(codes.Canceled, "buffer overflow") + case rst := <-sub.Out(): + if err := stream.Send(&spacemeshv2.ActivationHeader{ + Versioned: &spacemeshv2.ActivationHeader_V1{V1: toHeader(rst.VerifiedActivationTx)}}, + ); err != nil { + if errors.Is(err, io.EOF) { + return nil + } + return status.Error(codes.Internal, err.Error()) + } + } + } } func toAtx(atx *types.VerifiedActivationTx) *spacemeshv2.ActivationV1 { From 0c4328218ad32cf56beb77da5f96dc3f85c4908d Mon Sep 17 00:00:00 2001 From: Dmitry Date: Thu, 9 Nov 2023 12:29:01 +0100 Subject: [PATCH 09/12] track todo --- sql/atxs/atxs.go | 2 ++ 1 file changed, 2 insertions(+) diff --git a/sql/atxs/atxs.go b/sql/atxs/atxs.go index f74f18bae2..bdcdaaced1 100644 --- a/sql/atxs/atxs.go +++ b/sql/atxs/atxs.go @@ -471,6 +471,8 @@ func IterateAtxs(db sql.Executor, from, to types.EpochID, fn func(*types.Verifie return derr } +// TODO(dshulyak) extract code for query building into separate module + type token string const ( From 2da36346d11f7a772245bfc016edff38973f5aa0 Mon Sep 17 00:00:00 2001 From: Dmitry Date: Thu, 9 Nov 2023 12:38:09 +0100 Subject: [PATCH 10/12] fix where --- sql/atxs/atxs.go | 13 +++++++------ 1 file changed, 7 insertions(+), 6 deletions(-) diff --git a/sql/atxs/atxs.go b/sql/atxs/atxs.go index bdcdaaced1..99bb6fa0e3 100644 --- a/sql/atxs/atxs.go +++ b/sql/atxs/atxs.go @@ -533,15 +533,16 @@ func IterateAtxsOps( } func filterFrom(operations Operations) string { - query := " " - if len(operations.Filter) > 0 { - query = "where " - } + // TODO(dshulyak) using string writer will be more efficient + query := "" for i, op := range operations.Filter { + if i == 0 { + query += " " + string(Where) + } if i != 0 { - query += " " + string(And) + " " + query += " " + string(And) } - query += string(op.Field) + " " + string(op.Token) + " ?" + strconv.Itoa(i+1) + query += " " + string(op.Field) + " " + string(op.Token) + " ?" + strconv.Itoa(i+1) } for i, op := range operations.Other { query += " " + string(op.Field) + " " + string(op.Token) + " ?" + strconv.Itoa(i+1+len(operations.Filter)) From 82fd29ded5ab393840d0eb50bf3d5f7d6c2fb065 Mon Sep 17 00:00:00 2001 From: Dmitry Date: Thu, 9 Nov 2023 12:53:29 +0100 Subject: [PATCH 11/12] refactor offset / limit --- api/grpcserver/v2/activation.go | 4 ++-- sql/atxs/atxs.go | 6 +++--- 2 files changed, 5 insertions(+), 5 deletions(-) diff --git a/api/grpcserver/v2/activation.go b/api/grpcserver/v2/activation.go index 44172c5a07..7f20bdabba 100644 --- a/api/grpcserver/v2/activation.go +++ b/api/grpcserver/v2/activation.go @@ -252,7 +252,7 @@ func (s *ActivationService) List( if request.Limit > 100 { return nil, status.Error(codes.InvalidArgument, "limit is capped at 100") } else if request.Limit == 0 { - return nil, status.Error(codes.InvalidArgument, "limit must be set to a value below 100") + return nil, status.Error(codes.InvalidArgument, "limit must be set to <= 100") } rst := make([]*spacemeshv2.Activation, 0, request.Limit) if err := atxs.IterateAtxsOps(s.db, ops, func(atx *types.VerifiedActivationTx) bool { @@ -275,7 +275,7 @@ func (s *ActivationService) ListHeaders( if request.Limit > 10000 { return nil, status.Error(codes.InvalidArgument, "limit is capped at 10000") } else if request.Limit == 0 { - return nil, status.Error(codes.InvalidArgument, "limit must be set to a value below 10000") + return nil, status.Error(codes.InvalidArgument, "limit must be set to <= 10000") } rst := make([]*spacemeshv2.ActivationHeader, 0, request.Limit) if err := atxs.IterateAtxsOps(s.db, ops, func(atx *types.VerifiedActivationTx) bool { diff --git a/sql/atxs/atxs.go b/sql/atxs/atxs.go index 99bb6fa0e3..0ae4d6244a 100644 --- a/sql/atxs/atxs.go +++ b/sql/atxs/atxs.go @@ -544,15 +544,15 @@ func filterFrom(operations Operations) string { } query += " " + string(op.Field) + " " + string(op.Token) + " ?" + strconv.Itoa(i+1) } - for i, op := range operations.Other { - query += " " + string(op.Field) + " " + string(op.Token) + " ?" + strconv.Itoa(i+1+len(operations.Filter)) + for _, op := range operations.Other { + query += fmt.Sprintf(" %s %v", string(op.Field), op.Value) } return query } func bindingsFrom(operations Operations) sql.Encoder { return func(stmt *sql.Statement) { - for i, op := range append(operations.Filter, operations.Other...) { + for i, op := range operations.Filter { switch value := op.Value.(type) { case int64: stmt.BindInt64(i+1, value) From af95f33d74d94c6a111124ee046bebbfdfcc1d2b Mon Sep 17 00:00:00 2001 From: Dmitry Date: Thu, 9 Nov 2023 13:06:19 +0100 Subject: [PATCH 12/12] debug --- api/grpcserver/v2/activation.go | 12 ++++++------ sql/atxs/atxs.go | 3 ++- 2 files changed, 8 insertions(+), 7 deletions(-) diff --git a/api/grpcserver/v2/activation.go b/api/grpcserver/v2/activation.go index 7f20bdabba..dfb31aa40e 100644 --- a/api/grpcserver/v2/activation.go +++ b/api/grpcserver/v2/activation.go @@ -341,17 +341,17 @@ func toOperations(filter *spacemeshv2.ActivationRequest) (atxs.Operations, error Value: int64(filter.EndEpoch), }) } - if filter.Offset != 0 { - ops.Other = append(ops.Other, atxs.Op{ - Field: atxs.Offset, - Value: int64(filter.Offset), - }) - } if filter.Limit != 0 { ops.Other = append(ops.Other, atxs.Op{ Field: atxs.Limit, Value: int64(filter.Limit), }) } + if filter.Offset != 0 { + ops.Other = append(ops.Other, atxs.Op{ + Field: atxs.Offset, + Value: int64(filter.Offset), + }) + } return ops, nil } diff --git a/sql/atxs/atxs.go b/sql/atxs/atxs.go index 0ae4d6244a..92b84ee8f8 100644 --- a/sql/atxs/atxs.go +++ b/sql/atxs/atxs.go @@ -517,7 +517,7 @@ func IterateAtxsOps( ) error { var derr error _, err := db.Exec( - fullQuery+filterFrom(operations)+" order by epoch asc, id", + fullQuery+filterFrom(operations), bindingsFrom(operations), decoder(func(atx *types.VerifiedActivationTx, err error) bool { if atx != nil { @@ -544,6 +544,7 @@ func filterFrom(operations Operations) string { } query += " " + string(op.Field) + " " + string(op.Token) + " ?" + strconv.Itoa(i+1) } + query += " order by epoch asc, id" for _, op := range operations.Other { query += fmt.Sprintf(" %s %v", string(op.Field), op.Value) }