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 new file mode 100644 index 0000000000..dfb31aa40e --- /dev/null +++ b/api/grpcserver/v2/activation.go @@ -0,0 +1,357 @@ +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" +) + +const ( + Activation = "activation_v2" + ActivationStream = "activation_stream_v2" +) + +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 (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 { + 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 { + 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()) + } + 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 { + 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 { + 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()) + } + 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 { + v1 := &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()), + } + 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 { + 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} +} + +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" +} + +func (s *ActivationService) List( + ctx context.Context, + request *spacemeshv2.ActivationRequest, +) (*spacemeshv2.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. + 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 <= 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 := 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 <= 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 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 + } + 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), + }) + } + 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/go.mod b/go.mod index b9f8b4c765..4ac813cfad 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.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 f12a9a7be0..b0bff4e41f 100644 --- a/go.sum +++ b/go.sum @@ -639,6 +639,16 @@ 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/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/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= diff --git a/node/node.go b/node/node.go index f8a48a1589..86a1a94dad 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" @@ -1321,6 +1322,10 @@ func (app *App) initService( 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) } diff --git a/sql/atxs/atxs.go b/sql/atxs/atxs.go index 2e4d548989..92b84ee8f8 100644 --- a/sql/atxs/atxs.go +++ b/sql/atxs/atxs.go @@ -2,6 +2,7 @@ package atxs import ( "fmt" + "strconv" "time" "github.com/spacemeshos/go-spacemesh/codec" @@ -469,3 +470,98 @@ 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 ( + 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" + Offset field = "offset" + Limit field = "limit" +) + +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, + operations Operations, + fn func(*types.VerifiedActivationTx) bool, +) error { + var derr error + _, err := db.Exec( + fullQuery+filterFrom(operations), + 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 filterFrom(operations Operations) string { + // 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(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) + } + return query +} + +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)) + } + } + } +}