diff --git a/api/grpcserver/config.go b/api/grpcserver/config.go index c5974b2bbf..d1ea971662 100644 --- a/api/grpcserver/config.go +++ b/api/grpcserver/config.go @@ -27,23 +27,25 @@ type Config struct { type Service = string const ( - Admin Service = "admin" - Debug Service = "debug" - GlobalState Service = "global" - Mesh Service = "mesh" - Transaction Service = "transaction" - Activation Service = "activation" - Smesher Service = "smesher" - Post Service = "post" - Node Service = "node" + Admin Service = "admin" + Debug Service = "debug" + GlobalState Service = "global" + Mesh Service = "mesh" + Transaction Service = "transaction" + Activation Service = "activation" + Smesher Service = "smesher" + Post Service = "post" + Node Service = "node" + ActivationV2Alpha1 Service = "activation_v2alpha1" + ActivationStreamV2Alpha1 Service = "activation_stream_v2alpha1" ) // DefaultConfig defines the default configuration options for api. func DefaultConfig() Config { return Config{ - PublicServices: []Service{GlobalState, Mesh, Transaction, Node, Activation}, + PublicServices: []Service{GlobalState, Mesh, Transaction, Node, Activation, ActivationV2Alpha1}, PublicListener: "0.0.0.0:9092", - PrivateServices: []Service{Admin, Smesher, Debug}, + PrivateServices: []Service{Admin, Smesher, Debug, ActivationStreamV2Alpha1}, PrivateListener: "127.0.0.1:9093", PostServices: []Service{Post}, PostListener: "127.0.0.1:9094", diff --git a/api/grpcserver/v2alpha1/activation.go b/api/grpcserver/v2alpha1/activation.go new file mode 100644 index 0000000000..d113a0c3fd --- /dev/null +++ b/api/grpcserver/v2alpha1/activation.go @@ -0,0 +1,357 @@ +package v2alpha1 + +import ( + "context" + "errors" + "fmt" + "io" + + "github.com/grpc-ecosystem/go-grpc-middleware/logging/zap/ctxzap" + "github.com/grpc-ecosystem/grpc-gateway/v2/runtime" + spacemeshv2alpha1 "github.com/spacemeshos/api/release/go/spacemesh/v2alpha1" + "go.uber.org/zap" + "google.golang.org/grpc" + "google.golang.org/grpc/codes" + "google.golang.org/grpc/metadata" + "google.golang.org/grpc/status" + + "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" + "github.com/spacemeshos/go-spacemesh/sql/builder" +) + +const ( + Activation = "activation_v2alpha1" + ActivationStream = "activation_stream_v2alpha1" +) + +func NewActivationStreamService(db sql.Executor) *ActivationStreamService { + return &ActivationStreamService{db: db} +} + +type ActivationStreamService struct { + db sql.Executor +} + +func (s *ActivationStreamService) RegisterService(server *grpc.Server) { + spacemeshv2alpha1.RegisterActivationStreamServiceServer(server, s) +} + +func (s *ActivationStreamService) RegisterHandlerService(mux *runtime.ServeMux) error { + return spacemeshv2alpha1.RegisterActivationStreamServiceHandlerServer(context.Background(), mux, s) +} + +func (s *ActivationStreamService) String() string { + return "ActivationStreamService" +} + +func (s *ActivationStreamService) Stream( + request *spacemeshv2alpha1.ActivationStreamRequest, + stream spacemeshv2alpha1.ActivationStreamService_StreamServer, +) error { + var sub *events.BufferedSubscription[events.ActivationTx] + if request.Watch { + matcher := resultsMatcher{request, stream.Context()} + var err error + sub, err = events.SubscribeMatched(matcher.match) + 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 { + 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(&spacemeshv2alpha1.Activation{Versioned: &spacemeshv2alpha1.Activation_V1{V1: toAtx(atx)}}) + return ierr == nil + }); err != nil { + return status.Error(codes.Internal, err.Error()) + } + 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(): + err := stream.Send(&spacemeshv2alpha1.Activation{ + Versioned: &spacemeshv2alpha1.Activation_V1{V1: toAtx(rst.VerifiedActivationTx)}, + }) + switch { + case errors.Is(err, io.EOF): + return nil + case err != nil: + return status.Error(codes.Internal, err.Error()) + } + } + } +} + +func toAtx(atx *types.VerifiedActivationTx) *spacemeshv2alpha1.ActivationV1 { + v1 := &spacemeshv2alpha1.ActivationV1{ + Id: atx.ID().Bytes(), + NodeId: atx.SmesherID.Bytes(), + Signature: atx.Signature.Bytes(), + PublishEpoch: atx.PublishEpoch.Uint32(), + Sequence: atx.Sequence, + PreviousAtx: atx.PrevATXID[:], + PositioningAtx: atx.PositioningATX[:], + Coinbase: atx.Coinbase.String(), + Units: atx.NumUnits, + BaseHeight: uint32(atx.BaseTickHeight()), + Ticks: uint32(atx.TickCount()), + } + if atx.CommitmentATX != nil { + v1.CommittmentAtx = atx.CommitmentATX.Bytes() + } + if atx.VRFNonce != nil { + v1.VrfPostIndex = &spacemeshv2alpha1.VRFPostIndex{ + Nonce: uint64(*atx.VRFNonce), + } + } + if atx.InitialPost != nil { + v1.InitialPost = &spacemeshv2alpha1.Post{ + Nonce: atx.InitialPost.Nonce, + Indices: atx.InitialPost.Indices, + Pow: atx.InitialPost.Pow, + } + } + + if atx.NIPost == nil { + panic(fmt.Sprintf("nil nipost for atx %s", atx.ShortString())) + } + + if atx.NIPost.Post == nil { + panic(fmt.Sprintf("nil nipost post for atx %s", atx.ShortString())) + } + + if atx.NIPost.PostMetadata == nil { + panic(fmt.Sprintf("nil nipost post metadata for atx %s", atx.ShortString())) + } + + nipost := atx.NIPost + v1.Post = &spacemeshv2alpha1.Post{ + Nonce: nipost.Post.Nonce, + Indices: nipost.Post.Indices, + Pow: nipost.Post.Pow, + } + + v1.PostMeta = &spacemeshv2alpha1.PostMeta{ + Challenge: nipost.PostMetadata.Challenge, + LabelsPerUnit: nipost.PostMetadata.LabelsPerUnit, + } + + v1.Membership = &spacemeshv2alpha1.PoetMembershipProof{ + ProofNodes: make([][]byte, len(nipost.Membership.Nodes)), + Leaf: nipost.Membership.LeafIndex, + } + + for i, node := range nipost.Membership.Nodes { + v1.Membership.ProofNodes[i] = node.Bytes() + } + + return v1 +} + +func NewActivationService(db sql.Executor) *ActivationService { + return &ActivationService{db: db} +} + +type ActivationService struct { + db sql.Executor +} + +func (s *ActivationService) RegisterService(server *grpc.Server) { + spacemeshv2alpha1.RegisterActivationServiceServer(server, s) +} + +func (s *ActivationService) RegisterHandlerService(mux *runtime.ServeMux) error { + return spacemeshv2alpha1.RegisterActivationServiceHandlerServer(context.Background(), mux, s) +} + +// String returns the service name. +func (s *ActivationService) String() string { + return "ActivationService" +} + +func (s *ActivationService) List( + ctx context.Context, + request *spacemeshv2alpha1.ActivationRequest, +) (*spacemeshv2alpha1.ActivationList, error) { + ops, err := toOperations(request) + if err != nil { + return nil, status.Error(codes.InvalidArgument, err.Error()) + } + // every full atx is ~1KB. 100 atxs is ~100KB. + switch { + case request.Limit > 100: + return nil, status.Error(codes.InvalidArgument, "limit is capped at 100") + case request.Limit == 0: + return nil, status.Error(codes.InvalidArgument, "limit must be set to <= 100") + } + rst := make([]*spacemeshv2alpha1.Activation, 0, request.Limit) + if err := atxs.IterateAtxsOps(s.db, ops, func(atx *types.VerifiedActivationTx) bool { + rst = append(rst, &spacemeshv2alpha1.Activation{Versioned: &spacemeshv2alpha1.Activation_V1{V1: toAtx(atx)}}) + return true + }); err != nil { + return nil, status.Error(codes.Internal, err.Error()) + } + return &spacemeshv2alpha1.ActivationList{Activations: rst}, nil +} + +func (s *ActivationService) ActivationsCount( + ctx context.Context, + request *spacemeshv2alpha1.ActivationsCountRequest, +) (*spacemeshv2alpha1.ActivationsCountResponse, error) { + ops := builder.Operations{Filter: []builder.Op{ + { + Field: builder.Epoch, + Token: builder.Eq, + Value: int64(request.Epoch), + }, + }} + + count, err := atxs.CountAtxsByOps(s.db, ops) + if err != nil { + return nil, status.Error(codes.Internal, err.Error()) + } + + return &spacemeshv2alpha1.ActivationsCountResponse{Count: count}, nil +} + +func toRequest(filter *spacemeshv2alpha1.ActivationStreamRequest) *spacemeshv2alpha1.ActivationRequest { + return &spacemeshv2alpha1.ActivationRequest{ + NodeId: filter.NodeId, + Id: filter.Id, + Coinbase: filter.Coinbase, + StartEpoch: filter.StartEpoch, + EndEpoch: filter.EndEpoch, + } +} + +func toOperations(filter *spacemeshv2alpha1.ActivationRequest) (builder.Operations, error) { + ops := builder.Operations{} + if filter == nil { + return ops, nil + } + if filter.NodeId != nil { + ops.Filter = append(ops.Filter, builder.Op{ + Field: builder.Smesher, + Token: builder.Eq, + Value: filter.NodeId, + }) + } + if filter.Id != nil { + ops.Filter = append(ops.Filter, builder.Op{ + Field: builder.Id, + Token: builder.Eq, + Value: filter.Id, + }) + } + if len(filter.Coinbase) > 0 { + addr, err := types.StringToAddress(filter.Coinbase) + if err != nil { + return builder.Operations{}, err + } + ops.Filter = append(ops.Filter, builder.Op{ + Field: builder.Coinbase, + Token: builder.Eq, + Value: addr.Bytes(), + }) + } + if filter.StartEpoch != 0 { + ops.Filter = append(ops.Filter, builder.Op{ + Field: builder.Epoch, + Token: builder.Gte, + Value: int64(filter.StartEpoch), + }) + } + if filter.EndEpoch != 0 { + ops.Filter = append(ops.Filter, builder.Op{ + Field: builder.Epoch, + Token: builder.Lte, + Value: int64(filter.EndEpoch), + }) + } + + ops.Modifiers = append(ops.Modifiers, builder.Modifier{ + Key: builder.OrderBy, + Value: "epoch asc, id", + }) + + if filter.Limit != 0 { + ops.Modifiers = append(ops.Modifiers, builder.Modifier{ + Key: builder.Limit, + Value: int64(filter.Limit), + }) + } + if filter.Offset != 0 { + ops.Modifiers = append(ops.Modifiers, builder.Modifier{ + Key: builder.Offset, + Value: int64(filter.Offset), + }) + } + + return ops, nil +} + +type resultsMatcher struct { + *spacemeshv2alpha1.ActivationStreamRequest + ctx context.Context +} + +func (m *resultsMatcher) match(t *events.ActivationTx) bool { + if len(m.NodeId) > 0 { + var nodeId types.NodeID + copy(nodeId[:], m.NodeId) + + if t.SmesherID != nodeId { + return false + } + } + + if len(m.Id) > 0 { + var atxId types.ATXID + copy(atxId[:], m.Id) + + if t.ID() != atxId { + return false + } + } + + if len(m.Coinbase) > 0 { + addr, err := types.StringToAddress(m.Coinbase) + if err != nil { + ctxzap.Error(m.ctx, "unable to convert atx coinbase", zap.Error(err)) + return false + } + if t.Coinbase != addr { + return false + } + } + + if m.StartEpoch != 0 { + if t.PublishEpoch.Uint32() < m.StartEpoch { + return false + } + } + + if m.EndEpoch != 0 { + if t.PublishEpoch.Uint32() > m.EndEpoch { + return false + } + } + + return true +} diff --git a/api/grpcserver/v2alpha1/activation_test.go b/api/grpcserver/v2alpha1/activation_test.go new file mode 100644 index 0000000000..bb9d89cff0 --- /dev/null +++ b/api/grpcserver/v2alpha1/activation_test.go @@ -0,0 +1,235 @@ +package v2alpha1 + +import ( + "context" + "errors" + "io" + "testing" + + spacemeshv2alpha1 "github.com/spacemeshos/api/release/go/spacemesh/v2alpha1" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + "google.golang.org/grpc/codes" + "google.golang.org/grpc/status" + + "github.com/spacemeshos/go-spacemesh/common/fixture" + "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" +) + +func TestActivationService_List(t *testing.T) { + db := sql.InMemory() + ctx := context.Background() + + gen := fixture.NewAtxsGenerator() + activations := make([]types.VerifiedActivationTx, 100) + for i := range activations { + atx := gen.Next() + require.NoError(t, atxs.Add(db, atx)) + activations[i] = *atx + } + + svc := NewActivationService(db) + cfg, cleanup := launchServer(t, svc) + t.Cleanup(cleanup) + + conn := dialGrpc(ctx, t, cfg) + client := spacemeshv2alpha1.NewActivationServiceClient(conn) + + t.Run("limit set too high", func(t *testing.T) { + _, err := client.List(ctx, &spacemeshv2alpha1.ActivationRequest{Limit: 200}) + require.Error(t, err) + + s, ok := status.FromError(err) + require.True(t, ok) + assert.Equal(t, codes.InvalidArgument, s.Code()) + require.Equal(t, s.Message(), "limit is capped at 100") + }) + + t.Run("no limit set", func(t *testing.T) { + _, err := client.List(ctx, &spacemeshv2alpha1.ActivationRequest{}) + require.Error(t, err) + + s, ok := status.FromError(err) + require.True(t, ok) + assert.Equal(t, codes.InvalidArgument, s.Code()) + require.Equal(t, s.Message(), "limit must be set to <= 100") + }) + + t.Run("limit and offset", func(t *testing.T) { + list, err := client.List(ctx, &spacemeshv2alpha1.ActivationRequest{ + Limit: 25, + Offset: 50, + }) + require.NoError(t, err) + require.Len(t, list.Activations, 25) + }) + + t.Run("all", func(t *testing.T) { + list, err := client.List(ctx, &spacemeshv2alpha1.ActivationRequest{Limit: 100}) + require.NoError(t, err) + require.Equal(t, len(activations), len(list.Activations)) + }) + + t.Run("coinbase", func(t *testing.T) { + list, err := client.List(ctx, &spacemeshv2alpha1.ActivationRequest{ + Limit: 1, + Coinbase: activations[3].Coinbase.String(), + }) + require.NoError(t, err) + require.Equal(t, activations[3].ID().Bytes(), list.GetActivations()[0].GetV1().GetId()) + }) + + t.Run("nodeId", func(t *testing.T) { + list, err := client.List(ctx, &spacemeshv2alpha1.ActivationRequest{ + Limit: 1, + NodeId: activations[1].SmesherID.Bytes(), + }) + require.NoError(t, err) + require.Equal(t, activations[1].ID().Bytes(), list.GetActivations()[0].GetV1().GetId()) + }) + + t.Run("id", func(t *testing.T) { + list, err := client.List(ctx, &spacemeshv2alpha1.ActivationRequest{ + Limit: 1, + Id: activations[3].ID().Bytes(), + }) + require.NoError(t, err) + require.Equal(t, activations[3].ID().Bytes(), list.GetActivations()[0].GetV1().GetId()) + }) +} + +func TestActivationStreamService_Stream(t *testing.T) { + db := sql.InMemory() + ctx := context.Background() + + gen := fixture.NewAtxsGenerator() + activations := make([]types.VerifiedActivationTx, 100) + for i := range activations { + atx := gen.Next() + require.NoError(t, atxs.Add(db, atx)) + activations[i] = *atx + } + + svc := NewActivationStreamService(db) + cfg, cleanup := launchServer(t, svc) + t.Cleanup(cleanup) + + conn := dialGrpc(ctx, t, cfg) + client := spacemeshv2alpha1.NewActivationStreamServiceClient(conn) + + t.Run("all", func(t *testing.T) { + events.InitializeReporter() + t.Cleanup(events.CloseEventReporter) + + stream, err := client.Stream(ctx, &spacemeshv2alpha1.ActivationStreamRequest{}) + require.NoError(t, err) + + var i int + for { + _, err := stream.Recv() + if errors.Is(err, io.EOF) { + break + } + i++ + } + require.Len(t, activations, i) + }) + + t.Run("watch", func(t *testing.T) { + events.InitializeReporter() + t.Cleanup(events.CloseEventReporter) + + const ( + start = 100 + n = 10 + ) + + gen = fixture.NewAtxsGenerator().WithEpochs(start, 10) + var streamed []*events.ActivationTx + for i := 0; i < n; i++ { + streamed = append(streamed, &events.ActivationTx{VerifiedActivationTx: gen.Next()}) + } + + for _, tc := range []struct { + desc string + request *spacemeshv2alpha1.ActivationStreamRequest + }{ + { + desc: "ID", + request: &spacemeshv2alpha1.ActivationStreamRequest{ + Id: streamed[3].ID().Bytes(), + StartEpoch: start, + Watch: true, + }, + }, + { + desc: "NodeID", + request: &spacemeshv2alpha1.ActivationStreamRequest{ + NodeId: streamed[3].NodeID.Bytes(), + StartEpoch: start, + Watch: true, + }, + }, + { + desc: "Coinbase", + request: &spacemeshv2alpha1.ActivationStreamRequest{ + Coinbase: streamed[3].Coinbase.String(), + StartEpoch: start, + Watch: true, + }, + }, + } { + tc := tc + t.Run(tc.desc, func(t *testing.T) { + stream, err := client.Stream(ctx, tc.request) + require.NoError(t, err) + _, err = stream.Header() + require.NoError(t, err) + + var expect []*types.VerifiedActivationTx + for _, rst := range streamed { + events.ReportNewActivation(rst.VerifiedActivationTx) + matcher := resultsMatcher{tc.request, ctx} + if matcher.match(rst) { + expect = append(expect, rst.VerifiedActivationTx) + } + } + + for _, rst := range expect { + received, err := stream.Recv() + require.NoError(t, err) + require.Equal(t, toAtx(rst).String(), received.GetV1().String()) + } + }) + } + }) +} + +func TestActivationService_ActivationsCount(t *testing.T) { + db := sql.InMemory() + ctx := context.Background() + + gen := fixture.NewAtxsGenerator().WithEpochs(0, 1) + activations := make([]types.VerifiedActivationTx, 30) + for i := range activations { + atx := gen.Next() + require.NoError(t, atxs.Add(db, atx)) + activations[i] = *atx + } + + svc := NewActivationService(db) + cfg, cleanup := launchServer(t, svc) + t.Cleanup(cleanup) + + conn := dialGrpc(ctx, t, cfg) + client := spacemeshv2alpha1.NewActivationServiceClient(conn) + + count, err := client.ActivationsCount(ctx, &spacemeshv2alpha1.ActivationsCountRequest{ + Epoch: activations[3].PublishEpoch.Uint32(), + }) + require.NoError(t, err) + require.Len(t, activations, int(count.Count)) +} diff --git a/api/grpcserver/v2alpha1/v2alpha1_test.go b/api/grpcserver/v2alpha1/v2alpha1_test.go new file mode 100644 index 0000000000..d0010e8511 --- /dev/null +++ b/api/grpcserver/v2alpha1/v2alpha1_test.go @@ -0,0 +1,40 @@ +package v2alpha1 + +import ( + "context" + "testing" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + "go.uber.org/zap/zaptest" + "google.golang.org/grpc" + "google.golang.org/grpc/credentials/insecure" + + "github.com/spacemeshos/go-spacemesh/api/grpcserver" +) + +func launchServer(tb testing.TB, services ...grpcserver.ServiceAPI) (grpcserver.Config, func()) { + cfg := grpcserver.DefaultTestConfig() + grpcService, err := grpcserver.NewWithServices(cfg.PublicListener, zaptest.NewLogger(tb).Named("grpc"), cfg, services) + require.NoError(tb, err) + + // start gRPC server + require.NoError(tb, grpcService.Start()) + + // update config with bound addresses + cfg.PublicListener = grpcService.BoundAddress + + return cfg, func() { assert.NoError(tb, grpcService.Close()) } +} + +func dialGrpc(ctx context.Context, tb testing.TB, cfg grpcserver.Config) *grpc.ClientConn { + tb.Helper() + conn, err := grpc.DialContext(ctx, + cfg.PublicListener, + grpc.WithTransportCredentials(insecure.NewCredentials()), + grpc.WithBlock(), + ) + require.NoError(tb, err) + tb.Cleanup(func() { require.NoError(tb, conn.Close()) }) + return conn +} diff --git a/common/fixture/atxs.go b/common/fixture/atxs.go new file mode 100644 index 0000000000..727428d5f4 --- /dev/null +++ b/common/fixture/atxs.go @@ -0,0 +1,120 @@ +package fixture + +import ( + "math/rand" + "time" + + "github.com/spacemeshos/merkle-tree" + "github.com/spacemeshos/poet/shared" + + "github.com/spacemeshos/go-spacemesh/common/types" + "github.com/spacemeshos/go-spacemesh/genvm/sdk/wallet" + "github.com/spacemeshos/go-spacemesh/signing" +) + +// NewAtxsGenerator with some random parameters. +func NewAtxsGenerator() *AtxsGenerator { + return new(AtxsGenerator). + WithSeed(time.Now().UnixNano()). + WithEpochs(0, 10) +} + +// AtxsGenerator generates random activations. +// Activations are not syntactically or contextually valid. This is for testing databases and APIs. +type AtxsGenerator struct { + rng *rand.Rand + + Epochs []types.EpochID + Addrs []types.Address +} + +// WithSeed update randomness source. +func (g *AtxsGenerator) WithSeed(seed int64) *AtxsGenerator { + g.rng = rand.New(rand.NewSource(seed)) + return g +} + +// WithEpochs update epochs ids. +func (g *AtxsGenerator) WithEpochs(start, n int) *AtxsGenerator { + g.Epochs = nil + for i := 0; i < n; i++ { + g.Epochs = append(g.Epochs, types.EpochID(start+i)) + } + return g +} + +func (g *AtxsGenerator) newNIPost() *types.NIPost { + challenge := types.HexToHash32("55555") + poetRef := []byte("66666") + tree, err := merkle.NewTreeBuilder(). + WithHashFunc(shared.HashMembershipTreeNode). + WithLeavesToProve(map[uint64]bool{0: true}). + Build() + if err != nil { + panic("failed to add leaf to tree") + } + if err := tree.AddLeaf(challenge[:]); err != nil { + panic("failed to add leaf to tree") + } + nodes := tree.Proof() + nodesH32 := make([]types.Hash32, 0, len(nodes)) + for _, n := range nodes { + nodesH32 = append(nodesH32, types.BytesToHash(n)) + } + return &types.NIPost{ + Membership: types.MerkleProof{ + Nodes: nodesH32, + }, + Post: &types.Post{ + Nonce: 0, + Indices: []byte(nil), + }, + PostMetadata: &types.PostMetadata{ + Challenge: poetRef, + LabelsPerUnit: 2048, + }, + } +} + +// Next generates VerifiedActivationTx. +func (g *AtxsGenerator) Next() *types.VerifiedActivationTx { + var atx types.VerifiedActivationTx + + var prevAtxId types.ATXID + g.rng.Read(prevAtxId[:]) + var posAtxId types.ATXID + g.rng.Read(posAtxId[:]) + var nodeId types.NodeID + g.rng.Read(nodeId[:]) + + signer, err := signing.NewEdSigner() + if err != nil { + panic("failed to create signer") + } + + atx = types.VerifiedActivationTx{ + ActivationTx: &types.ActivationTx{ + InnerActivationTx: types.InnerActivationTx{ + NIPostChallenge: types.NIPostChallenge{ + Sequence: g.rng.Uint64(), + PrevATXID: prevAtxId, + PublishEpoch: g.Epochs[g.rng.Intn(len(g.Epochs))], + PositioningATX: posAtxId, + }, + Coinbase: wallet.Address(signer.PublicKey().Bytes()), + NumUnits: g.rng.Uint32(), + NodeID: &nodeId, + NIPost: g.newNIPost(), + }, + SmesherID: nodeId, + }, + } + + atx.SetEffectiveNumUnits(atx.NumUnits) + + var atxId types.ATXID + g.rng.Read(atxId[:]) + atx.SetID(atxId) + + return &atx +} diff --git a/go.mod b/go.mod index 1151594cd6..69eec921be 100644 --- a/go.mod +++ b/go.mod @@ -36,7 +36,7 @@ require ( github.com/quic-go/quic-go v0.41.0 github.com/santhosh-tekuri/jsonschema/v5 v5.3.1 github.com/seehuhn/mt19937 v1.0.0 - github.com/spacemeshos/api/release/go v1.28.0 + github.com/spacemeshos/api/release/go v1.28.1-0.20240215101325-80fe2752264b github.com/spacemeshos/economics v0.1.2 github.com/spacemeshos/fixed v0.1.1 github.com/spacemeshos/go-scale v1.1.12 diff --git a/go.sum b/go.sum index 66f56836c6..8de446a468 100644 --- a/go.sum +++ b/go.sum @@ -622,8 +622,8 @@ github.com/sourcegraph/annotate v0.0.0-20160123013949-f4cad6c6324d/go.mod h1:Udh github.com/sourcegraph/conc v0.3.0 h1:OQTbbt6P72L20UqAkXXuLOj79LfEanQ+YQFNpLA9ySo= github.com/sourcegraph/conc v0.3.0/go.mod h1:Sdozi7LEKbFPqYX2/J+iBAM6HpqSLTASQIKqDmF7Mt0= github.com/sourcegraph/syntaxhighlight v0.0.0-20170531221838-bd320f5d308e/go.mod h1:HuIsMU8RRBOtsCgI77wP899iHVBQpCmg4ErYMZB+2IA= -github.com/spacemeshos/api/release/go v1.28.0 h1:HmrNf0kV7U9o2rwqrmZ0tfPgQA8JQWY8EYGoiCB4bnI= -github.com/spacemeshos/api/release/go v1.28.0/go.mod h1:fK9RBD8eTVXHrqkkal2bwQB4N8M9sOhPs4rnVmWqEc0= +github.com/spacemeshos/api/release/go v1.28.1-0.20240215101325-80fe2752264b h1:SYHzzlFfoJwJeeDh0WZ3/+hObt33mZPlkGTNU2hD5Ds= +github.com/spacemeshos/api/release/go v1.28.1-0.20240215101325-80fe2752264b/go.mod h1:fK9RBD8eTVXHrqkkal2bwQB4N8M9sOhPs4rnVmWqEc0= github.com/spacemeshos/economics v0.1.2 h1:kw8cE5SMa/7svHOGorCd2w8ef1y8iP0p47/2VDOK8Ns= github.com/spacemeshos/economics v0.1.2/go.mod h1:ngeWn5E/jy9dJP1MHyuk3ehF8NBMTYhchqVDhAHUUNk= github.com/spacemeshos/fixed v0.1.1 h1:N1y4SUpq1EV+IdJrWJwUCt1oBFzeru/VKVcBsvPc2Fk= diff --git a/node/node.go b/node/node.go index 07650000b2..7e28f77a4f 100644 --- a/node/node.go +++ b/node/node.go @@ -34,6 +34,7 @@ import ( "github.com/spacemeshos/go-spacemesh/activation" "github.com/spacemeshos/go-spacemesh/api/grpcserver" + "github.com/spacemeshos/go-spacemesh/api/grpcserver/v2alpha1" "github.com/spacemeshos/go-spacemesh/atxsdata" "github.com/spacemeshos/go-spacemesh/beacon" "github.com/spacemeshos/go-spacemesh/blocks" @@ -1363,6 +1364,14 @@ func (app *App) grpcService(svc grpcserver.Service, lg log.Log) (grpcserver.Serv service := grpcserver.NewActivationService(app.cachedDB, types.ATXID(app.Config.Genesis.GoldenATX())) app.grpcServices[svc] = service return service, nil + case v2alpha1.Activation: + service := v2alpha1.NewActivationService(app.db) + app.grpcServices[svc] = service + return service, nil + case v2alpha1.ActivationStream: + service := v2alpha1.NewActivationStreamService(app.db) + app.grpcServices[svc] = service + return service, nil } return nil, fmt.Errorf("unknown service %s", svc) } diff --git a/sql/atxs/atxs.go b/sql/atxs/atxs.go index f19bfc1ead..adb17ade93 100644 --- a/sql/atxs/atxs.go +++ b/sql/atxs/atxs.go @@ -7,6 +7,7 @@ import ( "github.com/spacemeshos/go-spacemesh/codec" "github.com/spacemeshos/go-spacemesh/common/types" "github.com/spacemeshos/go-spacemesh/sql" + "github.com/spacemeshos/go-spacemesh/sql/builder" ) const fullQuery = `select id, atx, base_tick_height, tick_count, pubkey, @@ -504,3 +505,37 @@ func SetValidity(db sql.Executor, id types.ATXID, validity types.Validity) error } return nil } + +func IterateAtxsOps( + db sql.Executor, + operations builder.Operations, + fn func(*types.VerifiedActivationTx) bool, +) error { + var derr error + _, err := db.Exec( + fullQuery+builder.FilterFrom(operations), + builder.BindingsFrom(operations), + decoder(func(atx *types.VerifiedActivationTx, err error) bool { + if atx != nil { + return fn(atx) + } + derr = err + return derr == nil + })) + if err != nil { + return err + } + return derr +} + +func CountAtxsByOps(db sql.Executor, operations builder.Operations) (count uint32, err error) { + _, err = db.Exec( + "SELECT count(*) FROM atxs"+builder.FilterFrom(operations), + builder.BindingsFrom(operations), + func(stmt *sql.Statement) bool { + count = uint32(stmt.ColumnInt32(0)) + return true + }, + ) + return +} diff --git a/sql/builder/builder.go b/sql/builder/builder.go new file mode 100644 index 0000000000..e1f1303afd --- /dev/null +++ b/sql/builder/builder.go @@ -0,0 +1,91 @@ +package builder + +import ( + "fmt" + "strconv" + "strings" + + "github.com/spacemeshos/go-spacemesh/sql" +) + +type token string + +const ( + Eq token = "=" + NotEq token = "!=" + Gt token = ">" + Gte token = ">=" + Lt token = "<" + Lte token = "<=" +) + +type field string + +const ( + Epoch field = "epoch" + Smesher field = "pubkey" + Coinbase field = "coinbase" + Id field = "id" +) + +type modifier string + +const ( + Offset modifier = "offset" + Limit modifier = "limit" + OrderBy modifier = "order by" +) + +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 Modifier struct { + Key modifier + // Value will be type casted to one the expected types. + // Modifier will panic if it doesn't match any of expected. + Value any +} + +type Operations struct { + Filter []Op + Modifiers []Modifier +} + +func FilterFrom(operations Operations) string { + var queryBuilder strings.Builder + + for i, op := range operations.Filter { + if i == 0 { + queryBuilder.WriteString(" where") + } else { + queryBuilder.WriteString(" and") + } + queryBuilder.WriteString(" " + string(op.Field) + " " + string(op.Token) + " ?" + strconv.Itoa(i+1)) + } + + for _, m := range operations.Modifiers { + queryBuilder.WriteString(fmt.Sprintf(" %s %v", string(m.Key), m.Value)) + } + + return queryBuilder.String() +} + +func BindingsFrom(operations Operations) sql.Encoder { + return func(stmt *sql.Statement) { + for i, op := range operations.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)) + } + } + } +}