diff --git a/.golangci.yml b/.golangci.yml index 5648f715f..3c952f576 100644 --- a/.golangci.yml +++ b/.golangci.yml @@ -42,8 +42,16 @@ linters: - 140 - name: cognitive-complexity arguments: - - 20 + - 100 disabled: false + - name: cyclomatic + arguments: + - 100 + disabled: false + - name: function-length + arguments: + - 100 + - 200 exclusions: generated: lax presets: diff --git a/Makefile b/Makefile index 9b9cd8e01..25921ff10 100644 --- a/Makefile +++ b/Makefile @@ -4,7 +4,7 @@ clone-injective-indexer: git clone https://github.com/InjectiveLabs/injective-indexer.git -b v1.20.89 --depth 1 --single-branch clone-injective-core: - git clone https://github.com/InjectiveLabs/injective-core.git -b master --depth 1 --single-branch + git clone https://github.com/InjectiveLabs/injective-core.git -b ic-1047/data-pipeline --depth 1 --single-branch copy-exchange-client: clone-injective-indexer rm -rf exchange/* @@ -126,6 +126,8 @@ copy-chain-types: clone-injective-core cp injective-core/injective-chain/modules/wasmx/types/msgs.go chain/wasmx/types && \ cp injective-core/injective-chain/modules/wasmx/types/params.go chain/wasmx/types && \ cp injective-core/injective-chain/modules/wasmx/types/proposal.go chain/wasmx/types + mkdir -p chain/mq/types && \ + cp injective-core/injective-chain/mq/types/*.pb.go chain/mq/types mkdir -p chain/stream/types && \ cp injective-core/injective-chain/stream/types/*.pb.go chain/stream/types mkdir -p chain/stream/types/v2 && \ @@ -168,6 +170,16 @@ extract-message-names: update-ofac-list: go run examples/chain/ofac/1_DownloadOfacList/example.go +GO_INSTALL_BINDIR := $(shell go env GOBIN) +ifeq ($(GO_INSTALL_BINDIR),) +GO_INSTALL_BINDIR := $(shell go env GOPATH)/bin +endif + +install-mq: + mkdir -p $(GO_INSTALL_BINDIR) + go build -o $(GO_INSTALL_BINDIR)/mq-stream ./client/mq/stream/server + go build -o $(GO_INSTALL_BINDIR)/mq-detector ./client/mq/detector + tests: go clean -testcache && go test -race ./client/... ./ethereum/... coverage: @@ -189,4 +201,4 @@ lint-all: export GOPROXY=direct lint-all: golangci-lint run --timeout=15m -v -.PHONY: copy-exchange-client update-ofac-list tests coverage lint lint-last-commit lint-master lint-all extract-message-names +.PHONY: copy-exchange-client update-ofac-list install-mq tests coverage lint lint-last-commit lint-master lint-all extract-message-names diff --git a/chain/codec/types/types.go b/chain/codec/types/types.go index da0784c04..ceb722f1a 100644 --- a/chain/codec/types/types.go +++ b/chain/codec/types/types.go @@ -10,7 +10,6 @@ import ( "github.com/cosmos/cosmos-sdk/x/auth/tx" "github.com/cosmos/gogoproto/proto" - "github.com/InjectiveLabs/sdk-go/chain/app/ante/eip712" injcodec "github.com/InjectiveLabs/sdk-go/chain/codec" ) @@ -29,7 +28,7 @@ func MakeEncodingConfig() EncodingConfig { encodingConfig := EncodingConfig{ InterfaceRegistry: interfaceRegistry, Codec: appCodec, - TxConfig: tx.NewTxConfig(appCodec, tx.DefaultSignModes, eip712.NewSignModeHandler(appCodec)), + TxConfig: tx.NewTxConfig(appCodec, tx.DefaultSignModes), Amino: cdc, } diff --git a/chain/mq/types/query.pb.go b/chain/mq/types/query.pb.go new file mode 100644 index 000000000..c38c105a2 --- /dev/null +++ b/chain/mq/types/query.pb.go @@ -0,0 +1,1541 @@ +// Code generated by protoc-gen-gogo. DO NOT EDIT. +// source: injective/mq/v1beta1/query.proto + +package types + +import ( + context "context" + fmt "fmt" + v1 "github.com/cometbft/cometbft/api/cometbft/abci/v1" + _ "github.com/cosmos/gogoproto/gogoproto" + grpc1 "github.com/cosmos/gogoproto/grpc" + proto "github.com/cosmos/gogoproto/proto" + grpc "google.golang.org/grpc" + codes "google.golang.org/grpc/codes" + status "google.golang.org/grpc/status" + io "io" + math "math" + math_bits "math/bits" +) + +// Reference imports to suppress errors if they are not otherwise used. +var _ = proto.Marshal +var _ = fmt.Errorf +var _ = math.Inf + +// This is a compile-time assertion to ensure that this generated file +// is compatible with the proto package it is being compiled against. +// A compilation error at this line likely means your copy of the +// proto package needs to be updated. +const _ = proto.GoGoProtoPackageIsVersion3 // please upgrade the proto package + +// todo: still needed? +type AdditionalDataTypes int32 + +const ( + AdditionalDataTypes_EVM AdditionalDataTypes = 0 +) + +var AdditionalDataTypes_name = map[int32]string{ + 0: "EVM", +} + +var AdditionalDataTypes_value = map[string]int32{ + "EVM": 0, +} + +func (x AdditionalDataTypes) String() string { + return proto.EnumName(AdditionalDataTypes_name, int32(x)) +} + +func (AdditionalDataTypes) EnumDescriptor() ([]byte, []int) { + return fileDescriptor_6a5d23ccd5761942, []int{0} +} + +type EventType int32 + +const ( + EventType_ABCI EventType = 0 + EventType_PUBLISH EventType = 1 +) + +var EventType_name = map[int32]string{ + 0: "ABCI", + 1: "PUBLISH", +} + +var EventType_value = map[string]int32{ + "ABCI": 0, + "PUBLISH": 1, +} + +func (x EventType) String() string { + return proto.EnumName(EventType_name, int32(x)) +} + +func (EventType) EnumDescriptor() ([]byte, []int) { + return fileDescriptor_6a5d23ccd5761942, []int{1} +} + +type EventStreamRequest struct { + // a unique Kafka partition for this particular client. Reuse the value to + // resume where you left off with receiving data + ConsumerId string `protobuf:"bytes,1,opt,name=consumer_id,json=consumerId,proto3" json:"consumer_id,omitempty"` + // topic of interest (raw, latest, guaranteed) + Topic string `protobuf:"bytes,2,opt,name=topic,proto3" json:"topic,omitempty"` +} + +func (m *EventStreamRequest) Reset() { *m = EventStreamRequest{} } +func (m *EventStreamRequest) String() string { return proto.CompactTextString(m) } +func (*EventStreamRequest) ProtoMessage() {} +func (*EventStreamRequest) Descriptor() ([]byte, []int) { + return fileDescriptor_6a5d23ccd5761942, []int{0} +} +func (m *EventStreamRequest) XXX_Unmarshal(b []byte) error { + return m.Unmarshal(b) +} +func (m *EventStreamRequest) XXX_Marshal(b []byte, deterministic bool) ([]byte, error) { + if deterministic { + return xxx_messageInfo_EventStreamRequest.Marshal(b, m, deterministic) + } else { + b = b[:cap(b)] + n, err := m.MarshalToSizedBuffer(b) + if err != nil { + return nil, err + } + return b[:n], nil + } +} +func (m *EventStreamRequest) XXX_Merge(src proto.Message) { + xxx_messageInfo_EventStreamRequest.Merge(m, src) +} +func (m *EventStreamRequest) XXX_Size() int { + return m.Size() +} +func (m *EventStreamRequest) XXX_DiscardUnknown() { + xxx_messageInfo_EventStreamRequest.DiscardUnknown(m) +} + +var xxx_messageInfo_EventStreamRequest proto.InternalMessageInfo + +func (m *EventStreamRequest) GetConsumerId() string { + if m != nil { + return m.ConsumerId + } + return "" +} + +func (m *EventStreamRequest) GetTopic() string { + if m != nil { + return m.Topic + } + return "" +} + +type EventStreamResponse struct { + BlockHeight int64 `protobuf:"varint,1,opt,name=block_height,json=blockHeight,proto3" json:"block_height,omitempty"` + AppHash []byte `protobuf:"bytes,2,opt,name=app_hash,json=appHash,proto3" json:"app_hash,omitempty"` + LastAppHash []byte `protobuf:"bytes,3,opt,name=last_app_hash,json=lastAppHash,proto3" json:"last_app_hash,omitempty"` + BlockEvents EventSet `protobuf:"bytes,4,opt,name=block_events,json=blockEvents,proto3" json:"block_events"` + TxEvents []EventSet `protobuf:"bytes,5,rep,name=tx_events,json=txEvents,proto3" json:"tx_events"` + AdditionalDataEntries []*AdditionalDataEntry `protobuf:"bytes,6,rep,name=additional_data_entries,json=additionalDataEntries,proto3" json:"additional_data_entries,omitempty"` +} + +func (m *EventStreamResponse) Reset() { *m = EventStreamResponse{} } +func (m *EventStreamResponse) String() string { return proto.CompactTextString(m) } +func (*EventStreamResponse) ProtoMessage() {} +func (*EventStreamResponse) Descriptor() ([]byte, []int) { + return fileDescriptor_6a5d23ccd5761942, []int{1} +} +func (m *EventStreamResponse) XXX_Unmarshal(b []byte) error { + return m.Unmarshal(b) +} +func (m *EventStreamResponse) XXX_Marshal(b []byte, deterministic bool) ([]byte, error) { + if deterministic { + return xxx_messageInfo_EventStreamResponse.Marshal(b, m, deterministic) + } else { + b = b[:cap(b)] + n, err := m.MarshalToSizedBuffer(b) + if err != nil { + return nil, err + } + return b[:n], nil + } +} +func (m *EventStreamResponse) XXX_Merge(src proto.Message) { + xxx_messageInfo_EventStreamResponse.Merge(m, src) +} +func (m *EventStreamResponse) XXX_Size() int { + return m.Size() +} +func (m *EventStreamResponse) XXX_DiscardUnknown() { + xxx_messageInfo_EventStreamResponse.DiscardUnknown(m) +} + +var xxx_messageInfo_EventStreamResponse proto.InternalMessageInfo + +func (m *EventStreamResponse) GetBlockHeight() int64 { + if m != nil { + return m.BlockHeight + } + return 0 +} + +func (m *EventStreamResponse) GetAppHash() []byte { + if m != nil { + return m.AppHash + } + return nil +} + +func (m *EventStreamResponse) GetLastAppHash() []byte { + if m != nil { + return m.LastAppHash + } + return nil +} + +func (m *EventStreamResponse) GetBlockEvents() EventSet { + if m != nil { + return m.BlockEvents + } + return EventSet{} +} + +func (m *EventStreamResponse) GetTxEvents() []EventSet { + if m != nil { + return m.TxEvents + } + return nil +} + +func (m *EventStreamResponse) GetAdditionalDataEntries() []*AdditionalDataEntry { + if m != nil { + return m.AdditionalDataEntries + } + return nil +} + +type AdditionalDataEntry struct { + Type AdditionalDataTypes `protobuf:"varint,1,opt,name=type,proto3,enum=injective.mq.v1beta1.AdditionalDataTypes" json:"type,omitempty"` + Data []byte `protobuf:"bytes,2,opt,name=data,proto3" json:"data,omitempty"` +} + +func (m *AdditionalDataEntry) Reset() { *m = AdditionalDataEntry{} } +func (m *AdditionalDataEntry) String() string { return proto.CompactTextString(m) } +func (*AdditionalDataEntry) ProtoMessage() {} +func (*AdditionalDataEntry) Descriptor() ([]byte, []int) { + return fileDescriptor_6a5d23ccd5761942, []int{2} +} +func (m *AdditionalDataEntry) XXX_Unmarshal(b []byte) error { + return m.Unmarshal(b) +} +func (m *AdditionalDataEntry) XXX_Marshal(b []byte, deterministic bool) ([]byte, error) { + if deterministic { + return xxx_messageInfo_AdditionalDataEntry.Marshal(b, m, deterministic) + } else { + b = b[:cap(b)] + n, err := m.MarshalToSizedBuffer(b) + if err != nil { + return nil, err + } + return b[:n], nil + } +} +func (m *AdditionalDataEntry) XXX_Merge(src proto.Message) { + xxx_messageInfo_AdditionalDataEntry.Merge(m, src) +} +func (m *AdditionalDataEntry) XXX_Size() int { + return m.Size() +} +func (m *AdditionalDataEntry) XXX_DiscardUnknown() { + xxx_messageInfo_AdditionalDataEntry.DiscardUnknown(m) +} + +var xxx_messageInfo_AdditionalDataEntry proto.InternalMessageInfo + +func (m *AdditionalDataEntry) GetType() AdditionalDataTypes { + if m != nil { + return m.Type + } + return AdditionalDataTypes_EVM +} + +func (m *AdditionalDataEntry) GetData() []byte { + if m != nil { + return m.Data + } + return nil +} + +type EventSet struct { + PublishedEvents [][]byte `protobuf:"bytes,1,rep,name=published_events,json=publishedEvents,proto3" json:"published_events,omitempty"` + TrueOrders []EventType `protobuf:"varint,2,rep,packed,name=true_orders,json=trueOrders,proto3,enum=injective.mq.v1beta1.EventType" json:"true_orders,omitempty"` + AbciEvents []v1.Event `protobuf:"bytes,3,rep,name=abci_events,json=abciEvents,proto3" json:"abci_events"` +} + +func (m *EventSet) Reset() { *m = EventSet{} } +func (m *EventSet) String() string { return proto.CompactTextString(m) } +func (*EventSet) ProtoMessage() {} +func (*EventSet) Descriptor() ([]byte, []int) { + return fileDescriptor_6a5d23ccd5761942, []int{3} +} +func (m *EventSet) XXX_Unmarshal(b []byte) error { + return m.Unmarshal(b) +} +func (m *EventSet) XXX_Marshal(b []byte, deterministic bool) ([]byte, error) { + if deterministic { + return xxx_messageInfo_EventSet.Marshal(b, m, deterministic) + } else { + b = b[:cap(b)] + n, err := m.MarshalToSizedBuffer(b) + if err != nil { + return nil, err + } + return b[:n], nil + } +} +func (m *EventSet) XXX_Merge(src proto.Message) { + xxx_messageInfo_EventSet.Merge(m, src) +} +func (m *EventSet) XXX_Size() int { + return m.Size() +} +func (m *EventSet) XXX_DiscardUnknown() { + xxx_messageInfo_EventSet.DiscardUnknown(m) +} + +var xxx_messageInfo_EventSet proto.InternalMessageInfo + +func (m *EventSet) GetPublishedEvents() [][]byte { + if m != nil { + return m.PublishedEvents + } + return nil +} + +func (m *EventSet) GetTrueOrders() []EventType { + if m != nil { + return m.TrueOrders + } + return nil +} + +func (m *EventSet) GetAbciEvents() []v1.Event { + if m != nil { + return m.AbciEvents + } + return nil +} + +func init() { + proto.RegisterEnum("injective.mq.v1beta1.AdditionalDataTypes", AdditionalDataTypes_name, AdditionalDataTypes_value) + proto.RegisterEnum("injective.mq.v1beta1.EventType", EventType_name, EventType_value) + proto.RegisterType((*EventStreamRequest)(nil), "injective.mq.v1beta1.EventStreamRequest") + proto.RegisterType((*EventStreamResponse)(nil), "injective.mq.v1beta1.EventStreamResponse") + proto.RegisterType((*AdditionalDataEntry)(nil), "injective.mq.v1beta1.AdditionalDataEntry") + proto.RegisterType((*EventSet)(nil), "injective.mq.v1beta1.EventSet") +} + +func init() { proto.RegisterFile("injective/mq/v1beta1/query.proto", fileDescriptor_6a5d23ccd5761942) } + +var fileDescriptor_6a5d23ccd5761942 = []byte{ + // 592 bytes of a gzipped FileDescriptorProto + 0x1f, 0x8b, 0x08, 0x00, 0x00, 0x00, 0x00, 0x00, 0x02, 0xff, 0x8c, 0x54, 0xdf, 0x4e, 0xd4, 0x4e, + 0x14, 0x6e, 0xe9, 0x02, 0xcb, 0x29, 0x3f, 0x7e, 0x9b, 0x01, 0xc3, 0x4a, 0x4c, 0xa9, 0xbd, 0x5a, + 0x48, 0x6c, 0x05, 0xaf, 0x35, 0xec, 0x2a, 0x91, 0x8d, 0x18, 0x4d, 0x51, 0x2f, 0xb8, 0x69, 0xa6, + 0xed, 0xb8, 0x1d, 0xdd, 0xed, 0x74, 0x3b, 0xd3, 0x0d, 0xbc, 0x85, 0x6f, 0xe3, 0x2b, 0x70, 0xc9, + 0xa5, 0x57, 0xc6, 0x40, 0xe2, 0x73, 0x98, 0x99, 0xfe, 0x09, 0x28, 0x21, 0x7b, 0xd7, 0xf9, 0xce, + 0x77, 0xbe, 0x73, 0xe6, 0x9b, 0x73, 0x0a, 0x36, 0x4d, 0xbf, 0x90, 0x48, 0xd0, 0x19, 0xf1, 0x26, + 0x53, 0x6f, 0xb6, 0x17, 0x12, 0x81, 0xf7, 0xbc, 0x69, 0x41, 0xf2, 0x73, 0x37, 0xcb, 0x99, 0x60, + 0x68, 0xa3, 0x61, 0xb8, 0x93, 0xa9, 0x5b, 0x31, 0xb6, 0x36, 0x46, 0x6c, 0xc4, 0x14, 0xc1, 0x93, + 0x5f, 0x25, 0x77, 0xeb, 0x51, 0xc4, 0x26, 0x44, 0x84, 0x9f, 0x85, 0x87, 0xc3, 0x88, 0x7a, 0xb3, + 0x3d, 0x4f, 0x9c, 0x67, 0x84, 0x97, 0x51, 0xe7, 0x0d, 0xa0, 0xc3, 0x19, 0x49, 0xc5, 0x89, 0xc8, + 0x09, 0x9e, 0xf8, 0x64, 0x5a, 0x10, 0x2e, 0xd0, 0x36, 0x98, 0x11, 0x4b, 0x79, 0x31, 0x21, 0x79, + 0x40, 0xe3, 0xae, 0x6e, 0xeb, 0xbd, 0x15, 0x1f, 0x6a, 0x68, 0x18, 0xa3, 0x0d, 0x58, 0x14, 0x2c, + 0xa3, 0x51, 0x77, 0x41, 0x85, 0xca, 0x83, 0xf3, 0x7b, 0x01, 0xd6, 0x6f, 0xa9, 0xf1, 0x8c, 0xa5, + 0x9c, 0xa0, 0xc7, 0xb0, 0x1a, 0x8e, 0x59, 0xf4, 0x35, 0x48, 0x08, 0x1d, 0x25, 0x42, 0xe9, 0x19, + 0xbe, 0xa9, 0xb0, 0x23, 0x05, 0xa1, 0x87, 0xd0, 0xc6, 0x59, 0x16, 0x24, 0x98, 0x27, 0x4a, 0x73, + 0xd5, 0x5f, 0xc6, 0x59, 0x76, 0x84, 0x79, 0x82, 0x1c, 0xf8, 0x6f, 0x8c, 0xb9, 0x08, 0x9a, 0xb8, + 0xa1, 0xe2, 0xa6, 0x04, 0xfb, 0x15, 0xe7, 0x75, 0x5d, 0x81, 0xc8, 0xf2, 0xbc, 0xdb, 0xb2, 0xf5, + 0x9e, 0xb9, 0x6f, 0xb9, 0x77, 0xf9, 0xe4, 0x96, 0x2d, 0x12, 0x31, 0x68, 0x5d, 0xfc, 0xdc, 0xd6, + 0xaa, 0x3e, 0x14, 0xc8, 0x51, 0x1f, 0x56, 0xc4, 0x59, 0xad, 0xb2, 0x68, 0x1b, 0x73, 0xab, 0xb4, + 0xc5, 0x59, 0x25, 0x81, 0x61, 0x13, 0xc7, 0x31, 0x15, 0x94, 0xa5, 0x78, 0x1c, 0xc4, 0x58, 0xe0, + 0x80, 0xa4, 0x22, 0xa7, 0x84, 0x77, 0x97, 0x94, 0xe0, 0xce, 0xdd, 0x82, 0xfd, 0x26, 0xe9, 0x15, + 0x16, 0xf8, 0x30, 0x15, 0xf9, 0xb9, 0xff, 0x00, 0xff, 0x03, 0x52, 0xc2, 0x9d, 0x04, 0xd6, 0xef, + 0x60, 0xa3, 0xe7, 0xd0, 0x92, 0x6f, 0xab, 0xfc, 0x5d, 0x9b, 0xaf, 0xcc, 0x07, 0x39, 0x0b, 0xbe, + 0x4a, 0x43, 0x08, 0x5a, 0xb2, 0xdb, 0xca, 0x7f, 0xf5, 0xed, 0x7c, 0xd7, 0xa1, 0x5d, 0xdf, 0x14, + 0xed, 0x40, 0x27, 0x2b, 0xc2, 0x31, 0xe5, 0x09, 0x89, 0x6b, 0x8f, 0x74, 0xdb, 0xe8, 0xad, 0xfa, + 0xff, 0x37, 0x78, 0x65, 0xc2, 0x01, 0x98, 0x22, 0x2f, 0x48, 0xc0, 0xf2, 0x98, 0xe4, 0xbc, 0xbb, + 0x60, 0x1b, 0xbd, 0xb5, 0xfd, 0xed, 0x7b, 0x9c, 0x94, 0x8d, 0xf8, 0x20, 0x73, 0xde, 0xa9, 0x14, + 0xf4, 0x02, 0x4c, 0x39, 0xb0, 0x75, 0x1d, 0x43, 0x59, 0xb7, 0xe9, 0xd6, 0xd3, 0xec, 0xca, 0xa0, + 0x3b, 0xab, 0xb2, 0xab, 0x47, 0x00, 0x09, 0x96, 0x1d, 0xec, 0x5a, 0x7f, 0x7b, 0xa4, 0xae, 0x8a, + 0x96, 0xc1, 0x38, 0xfc, 0xf4, 0xb6, 0xa3, 0xed, 0x3a, 0xb0, 0xd2, 0x14, 0x46, 0x6d, 0x68, 0xf5, + 0x07, 0x2f, 0x87, 0x1d, 0x0d, 0x99, 0xb0, 0xfc, 0xfe, 0xe3, 0xe0, 0x78, 0x78, 0x72, 0xd4, 0xd1, + 0xf7, 0x39, 0x98, 0x37, 0xe6, 0x19, 0xc5, 0xb7, 0x8f, 0xbd, 0xfb, 0x06, 0xe3, 0xe6, 0x3e, 0x6d, + 0xed, 0xcc, 0xc1, 0x2c, 0x77, 0xe5, 0xa9, 0x3e, 0x38, 0xbd, 0xb8, 0xb2, 0xf4, 0xcb, 0x2b, 0x4b, + 0xff, 0x75, 0x65, 0xe9, 0xdf, 0xae, 0x2d, 0xed, 0xf2, 0xda, 0xd2, 0x7e, 0x5c, 0x5b, 0xda, 0xe9, + 0xc1, 0x88, 0x8a, 0xa4, 0x08, 0xa5, 0x07, 0xde, 0xb0, 0x16, 0x3c, 0xc6, 0x21, 0xf7, 0x1a, 0xf9, + 0x27, 0x11, 0xcb, 0xc9, 0xcd, 0x63, 0x82, 0x69, 0x2a, 0x7f, 0x23, 0x6a, 0xe9, 0xc3, 0x25, 0xb5, + 0xf5, 0xcf, 0xfe, 0x04, 0x00, 0x00, 0xff, 0xff, 0x39, 0xcf, 0xe4, 0x6a, 0x63, 0x04, 0x00, 0x00, +} + +// Reference imports to suppress errors if they are not otherwise used. +var _ context.Context +var _ grpc.ClientConn + +// This is a compile-time assertion to ensure that this generated file +// is compatible with the grpc package it is being compiled against. +const _ = grpc.SupportPackageIsVersion4 + +// EventStreamClient is the client API for EventStream service. +// +// For semantics around ctx use and closing/ending streaming RPCs, please refer to https://godoc.org/google.golang.org/grpc#ClientConn.NewStream. +type EventStreamClient interface { + EventStream(ctx context.Context, in *EventStreamRequest, opts ...grpc.CallOption) (EventStream_EventStreamClient, error) +} + +type eventStreamClient struct { + cc grpc1.ClientConn +} + +func NewEventStreamClient(cc grpc1.ClientConn) EventStreamClient { + return &eventStreamClient{cc} +} + +func (c *eventStreamClient) EventStream(ctx context.Context, in *EventStreamRequest, opts ...grpc.CallOption) (EventStream_EventStreamClient, error) { + stream, err := c.cc.NewStream(ctx, &_EventStream_serviceDesc.Streams[0], "/injective.mq.v1beta1.EventStream/EventStream", opts...) + if err != nil { + return nil, err + } + x := &eventStreamEventStreamClient{stream} + if err := x.ClientStream.SendMsg(in); err != nil { + return nil, err + } + if err := x.ClientStream.CloseSend(); err != nil { + return nil, err + } + return x, nil +} + +type EventStream_EventStreamClient interface { + Recv() (*EventStreamResponse, error) + grpc.ClientStream +} + +type eventStreamEventStreamClient struct { + grpc.ClientStream +} + +func (x *eventStreamEventStreamClient) Recv() (*EventStreamResponse, error) { + m := new(EventStreamResponse) + if err := x.ClientStream.RecvMsg(m); err != nil { + return nil, err + } + return m, nil +} + +// EventStreamServer is the server API for EventStream service. +type EventStreamServer interface { + EventStream(*EventStreamRequest, EventStream_EventStreamServer) error +} + +// UnimplementedEventStreamServer can be embedded to have forward compatible implementations. +type UnimplementedEventStreamServer struct { +} + +func (*UnimplementedEventStreamServer) EventStream(req *EventStreamRequest, srv EventStream_EventStreamServer) error { + return status.Errorf(codes.Unimplemented, "method EventStream not implemented") +} + +func RegisterEventStreamServer(s grpc1.Server, srv EventStreamServer) { + s.RegisterService(&_EventStream_serviceDesc, srv) +} + +func _EventStream_EventStream_Handler(srv interface{}, stream grpc.ServerStream) error { + m := new(EventStreamRequest) + if err := stream.RecvMsg(m); err != nil { + return err + } + return srv.(EventStreamServer).EventStream(m, &eventStreamEventStreamServer{stream}) +} + +type EventStream_EventStreamServer interface { + Send(*EventStreamResponse) error + grpc.ServerStream +} + +type eventStreamEventStreamServer struct { + grpc.ServerStream +} + +func (x *eventStreamEventStreamServer) Send(m *EventStreamResponse) error { + return x.ServerStream.SendMsg(m) +} + +var _EventStream_serviceDesc = grpc.ServiceDesc{ + ServiceName: "injective.mq.v1beta1.EventStream", + HandlerType: (*EventStreamServer)(nil), + Methods: []grpc.MethodDesc{}, + Streams: []grpc.StreamDesc{ + { + StreamName: "EventStream", + Handler: _EventStream_EventStream_Handler, + ServerStreams: true, + }, + }, + Metadata: "injective/mq/v1beta1/query.proto", +} + +func (m *EventStreamRequest) Marshal() (dAtA []byte, err error) { + size := m.Size() + dAtA = make([]byte, size) + n, err := m.MarshalToSizedBuffer(dAtA[:size]) + if err != nil { + return nil, err + } + return dAtA[:n], nil +} + +func (m *EventStreamRequest) MarshalTo(dAtA []byte) (int, error) { + size := m.Size() + return m.MarshalToSizedBuffer(dAtA[:size]) +} + +func (m *EventStreamRequest) MarshalToSizedBuffer(dAtA []byte) (int, error) { + i := len(dAtA) + _ = i + var l int + _ = l + if len(m.Topic) > 0 { + i -= len(m.Topic) + copy(dAtA[i:], m.Topic) + i = encodeVarintQuery(dAtA, i, uint64(len(m.Topic))) + i-- + dAtA[i] = 0x12 + } + if len(m.ConsumerId) > 0 { + i -= len(m.ConsumerId) + copy(dAtA[i:], m.ConsumerId) + i = encodeVarintQuery(dAtA, i, uint64(len(m.ConsumerId))) + i-- + dAtA[i] = 0xa + } + return len(dAtA) - i, nil +} + +func (m *EventStreamResponse) Marshal() (dAtA []byte, err error) { + size := m.Size() + dAtA = make([]byte, size) + n, err := m.MarshalToSizedBuffer(dAtA[:size]) + if err != nil { + return nil, err + } + return dAtA[:n], nil +} + +func (m *EventStreamResponse) MarshalTo(dAtA []byte) (int, error) { + size := m.Size() + return m.MarshalToSizedBuffer(dAtA[:size]) +} + +func (m *EventStreamResponse) MarshalToSizedBuffer(dAtA []byte) (int, error) { + i := len(dAtA) + _ = i + var l int + _ = l + if len(m.AdditionalDataEntries) > 0 { + for iNdEx := len(m.AdditionalDataEntries) - 1; iNdEx >= 0; iNdEx-- { + { + size, err := m.AdditionalDataEntries[iNdEx].MarshalToSizedBuffer(dAtA[:i]) + if err != nil { + return 0, err + } + i -= size + i = encodeVarintQuery(dAtA, i, uint64(size)) + } + i-- + dAtA[i] = 0x32 + } + } + if len(m.TxEvents) > 0 { + for iNdEx := len(m.TxEvents) - 1; iNdEx >= 0; iNdEx-- { + { + size, err := m.TxEvents[iNdEx].MarshalToSizedBuffer(dAtA[:i]) + if err != nil { + return 0, err + } + i -= size + i = encodeVarintQuery(dAtA, i, uint64(size)) + } + i-- + dAtA[i] = 0x2a + } + } + { + size, err := m.BlockEvents.MarshalToSizedBuffer(dAtA[:i]) + if err != nil { + return 0, err + } + i -= size + i = encodeVarintQuery(dAtA, i, uint64(size)) + } + i-- + dAtA[i] = 0x22 + if len(m.LastAppHash) > 0 { + i -= len(m.LastAppHash) + copy(dAtA[i:], m.LastAppHash) + i = encodeVarintQuery(dAtA, i, uint64(len(m.LastAppHash))) + i-- + dAtA[i] = 0x1a + } + if len(m.AppHash) > 0 { + i -= len(m.AppHash) + copy(dAtA[i:], m.AppHash) + i = encodeVarintQuery(dAtA, i, uint64(len(m.AppHash))) + i-- + dAtA[i] = 0x12 + } + if m.BlockHeight != 0 { + i = encodeVarintQuery(dAtA, i, uint64(m.BlockHeight)) + i-- + dAtA[i] = 0x8 + } + return len(dAtA) - i, nil +} + +func (m *AdditionalDataEntry) Marshal() (dAtA []byte, err error) { + size := m.Size() + dAtA = make([]byte, size) + n, err := m.MarshalToSizedBuffer(dAtA[:size]) + if err != nil { + return nil, err + } + return dAtA[:n], nil +} + +func (m *AdditionalDataEntry) MarshalTo(dAtA []byte) (int, error) { + size := m.Size() + return m.MarshalToSizedBuffer(dAtA[:size]) +} + +func (m *AdditionalDataEntry) MarshalToSizedBuffer(dAtA []byte) (int, error) { + i := len(dAtA) + _ = i + var l int + _ = l + if len(m.Data) > 0 { + i -= len(m.Data) + copy(dAtA[i:], m.Data) + i = encodeVarintQuery(dAtA, i, uint64(len(m.Data))) + i-- + dAtA[i] = 0x12 + } + if m.Type != 0 { + i = encodeVarintQuery(dAtA, i, uint64(m.Type)) + i-- + dAtA[i] = 0x8 + } + return len(dAtA) - i, nil +} + +func (m *EventSet) Marshal() (dAtA []byte, err error) { + size := m.Size() + dAtA = make([]byte, size) + n, err := m.MarshalToSizedBuffer(dAtA[:size]) + if err != nil { + return nil, err + } + return dAtA[:n], nil +} + +func (m *EventSet) MarshalTo(dAtA []byte) (int, error) { + size := m.Size() + return m.MarshalToSizedBuffer(dAtA[:size]) +} + +func (m *EventSet) MarshalToSizedBuffer(dAtA []byte) (int, error) { + i := len(dAtA) + _ = i + var l int + _ = l + if len(m.AbciEvents) > 0 { + for iNdEx := len(m.AbciEvents) - 1; iNdEx >= 0; iNdEx-- { + { + size, err := m.AbciEvents[iNdEx].MarshalToSizedBuffer(dAtA[:i]) + if err != nil { + return 0, err + } + i -= size + i = encodeVarintQuery(dAtA, i, uint64(size)) + } + i-- + dAtA[i] = 0x1a + } + } + if len(m.TrueOrders) > 0 { + dAtA3 := make([]byte, len(m.TrueOrders)*10) + var j2 int + for _, num := range m.TrueOrders { + for num >= 1<<7 { + dAtA3[j2] = uint8(uint64(num)&0x7f | 0x80) + num >>= 7 + j2++ + } + dAtA3[j2] = uint8(num) + j2++ + } + i -= j2 + copy(dAtA[i:], dAtA3[:j2]) + i = encodeVarintQuery(dAtA, i, uint64(j2)) + i-- + dAtA[i] = 0x12 + } + if len(m.PublishedEvents) > 0 { + for iNdEx := len(m.PublishedEvents) - 1; iNdEx >= 0; iNdEx-- { + i -= len(m.PublishedEvents[iNdEx]) + copy(dAtA[i:], m.PublishedEvents[iNdEx]) + i = encodeVarintQuery(dAtA, i, uint64(len(m.PublishedEvents[iNdEx]))) + i-- + dAtA[i] = 0xa + } + } + return len(dAtA) - i, nil +} + +func encodeVarintQuery(dAtA []byte, offset int, v uint64) int { + offset -= sovQuery(v) + base := offset + for v >= 1<<7 { + dAtA[offset] = uint8(v&0x7f | 0x80) + v >>= 7 + offset++ + } + dAtA[offset] = uint8(v) + return base +} +func (m *EventStreamRequest) Size() (n int) { + if m == nil { + return 0 + } + var l int + _ = l + l = len(m.ConsumerId) + if l > 0 { + n += 1 + l + sovQuery(uint64(l)) + } + l = len(m.Topic) + if l > 0 { + n += 1 + l + sovQuery(uint64(l)) + } + return n +} + +func (m *EventStreamResponse) Size() (n int) { + if m == nil { + return 0 + } + var l int + _ = l + if m.BlockHeight != 0 { + n += 1 + sovQuery(uint64(m.BlockHeight)) + } + l = len(m.AppHash) + if l > 0 { + n += 1 + l + sovQuery(uint64(l)) + } + l = len(m.LastAppHash) + if l > 0 { + n += 1 + l + sovQuery(uint64(l)) + } + l = m.BlockEvents.Size() + n += 1 + l + sovQuery(uint64(l)) + if len(m.TxEvents) > 0 { + for _, e := range m.TxEvents { + l = e.Size() + n += 1 + l + sovQuery(uint64(l)) + } + } + if len(m.AdditionalDataEntries) > 0 { + for _, e := range m.AdditionalDataEntries { + l = e.Size() + n += 1 + l + sovQuery(uint64(l)) + } + } + return n +} + +func (m *AdditionalDataEntry) Size() (n int) { + if m == nil { + return 0 + } + var l int + _ = l + if m.Type != 0 { + n += 1 + sovQuery(uint64(m.Type)) + } + l = len(m.Data) + if l > 0 { + n += 1 + l + sovQuery(uint64(l)) + } + return n +} + +func (m *EventSet) Size() (n int) { + if m == nil { + return 0 + } + var l int + _ = l + if len(m.PublishedEvents) > 0 { + for _, b := range m.PublishedEvents { + l = len(b) + n += 1 + l + sovQuery(uint64(l)) + } + } + if len(m.TrueOrders) > 0 { + l = 0 + for _, e := range m.TrueOrders { + l += sovQuery(uint64(e)) + } + n += 1 + sovQuery(uint64(l)) + l + } + if len(m.AbciEvents) > 0 { + for _, e := range m.AbciEvents { + l = e.Size() + n += 1 + l + sovQuery(uint64(l)) + } + } + return n +} + +func sovQuery(x uint64) (n int) { + return (math_bits.Len64(x|1) + 6) / 7 +} +func sozQuery(x uint64) (n int) { + return sovQuery(uint64((x << 1) ^ uint64((int64(x) >> 63)))) +} +func (m *EventStreamRequest) Unmarshal(dAtA []byte) error { + l := len(dAtA) + iNdEx := 0 + for iNdEx < l { + preIndex := iNdEx + var wire uint64 + for shift := uint(0); ; shift += 7 { + if shift >= 64 { + return ErrIntOverflowQuery + } + if iNdEx >= l { + return io.ErrUnexpectedEOF + } + b := dAtA[iNdEx] + iNdEx++ + wire |= uint64(b&0x7F) << shift + if b < 0x80 { + break + } + } + fieldNum := int32(wire >> 3) + wireType := int(wire & 0x7) + if wireType == 4 { + return fmt.Errorf("proto: EventStreamRequest: wiretype end group for non-group") + } + if fieldNum <= 0 { + return fmt.Errorf("proto: EventStreamRequest: illegal tag %d (wire type %d)", fieldNum, wire) + } + switch fieldNum { + case 1: + if wireType != 2 { + return fmt.Errorf("proto: wrong wireType = %d for field ConsumerId", wireType) + } + var stringLen uint64 + for shift := uint(0); ; shift += 7 { + if shift >= 64 { + return ErrIntOverflowQuery + } + if iNdEx >= l { + return io.ErrUnexpectedEOF + } + b := dAtA[iNdEx] + iNdEx++ + stringLen |= uint64(b&0x7F) << shift + if b < 0x80 { + break + } + } + intStringLen := int(stringLen) + if intStringLen < 0 { + return ErrInvalidLengthQuery + } + postIndex := iNdEx + intStringLen + if postIndex < 0 { + return ErrInvalidLengthQuery + } + if postIndex > l { + return io.ErrUnexpectedEOF + } + m.ConsumerId = string(dAtA[iNdEx:postIndex]) + iNdEx = postIndex + case 2: + if wireType != 2 { + return fmt.Errorf("proto: wrong wireType = %d for field Topic", wireType) + } + var stringLen uint64 + for shift := uint(0); ; shift += 7 { + if shift >= 64 { + return ErrIntOverflowQuery + } + if iNdEx >= l { + return io.ErrUnexpectedEOF + } + b := dAtA[iNdEx] + iNdEx++ + stringLen |= uint64(b&0x7F) << shift + if b < 0x80 { + break + } + } + intStringLen := int(stringLen) + if intStringLen < 0 { + return ErrInvalidLengthQuery + } + postIndex := iNdEx + intStringLen + if postIndex < 0 { + return ErrInvalidLengthQuery + } + if postIndex > l { + return io.ErrUnexpectedEOF + } + m.Topic = string(dAtA[iNdEx:postIndex]) + iNdEx = postIndex + default: + iNdEx = preIndex + skippy, err := skipQuery(dAtA[iNdEx:]) + if err != nil { + return err + } + if (skippy < 0) || (iNdEx+skippy) < 0 { + return ErrInvalidLengthQuery + } + if (iNdEx + skippy) > l { + return io.ErrUnexpectedEOF + } + iNdEx += skippy + } + } + + if iNdEx > l { + return io.ErrUnexpectedEOF + } + return nil +} +func (m *EventStreamResponse) Unmarshal(dAtA []byte) error { + l := len(dAtA) + iNdEx := 0 + for iNdEx < l { + preIndex := iNdEx + var wire uint64 + for shift := uint(0); ; shift += 7 { + if shift >= 64 { + return ErrIntOverflowQuery + } + if iNdEx >= l { + return io.ErrUnexpectedEOF + } + b := dAtA[iNdEx] + iNdEx++ + wire |= uint64(b&0x7F) << shift + if b < 0x80 { + break + } + } + fieldNum := int32(wire >> 3) + wireType := int(wire & 0x7) + if wireType == 4 { + return fmt.Errorf("proto: EventStreamResponse: wiretype end group for non-group") + } + if fieldNum <= 0 { + return fmt.Errorf("proto: EventStreamResponse: illegal tag %d (wire type %d)", fieldNum, wire) + } + switch fieldNum { + case 1: + if wireType != 0 { + return fmt.Errorf("proto: wrong wireType = %d for field BlockHeight", wireType) + } + m.BlockHeight = 0 + for shift := uint(0); ; shift += 7 { + if shift >= 64 { + return ErrIntOverflowQuery + } + if iNdEx >= l { + return io.ErrUnexpectedEOF + } + b := dAtA[iNdEx] + iNdEx++ + m.BlockHeight |= int64(b&0x7F) << shift + if b < 0x80 { + break + } + } + case 2: + if wireType != 2 { + return fmt.Errorf("proto: wrong wireType = %d for field AppHash", wireType) + } + var byteLen int + for shift := uint(0); ; shift += 7 { + if shift >= 64 { + return ErrIntOverflowQuery + } + if iNdEx >= l { + return io.ErrUnexpectedEOF + } + b := dAtA[iNdEx] + iNdEx++ + byteLen |= int(b&0x7F) << shift + if b < 0x80 { + break + } + } + if byteLen < 0 { + return ErrInvalidLengthQuery + } + postIndex := iNdEx + byteLen + if postIndex < 0 { + return ErrInvalidLengthQuery + } + if postIndex > l { + return io.ErrUnexpectedEOF + } + m.AppHash = append(m.AppHash[:0], dAtA[iNdEx:postIndex]...) + if m.AppHash == nil { + m.AppHash = []byte{} + } + iNdEx = postIndex + case 3: + if wireType != 2 { + return fmt.Errorf("proto: wrong wireType = %d for field LastAppHash", wireType) + } + var byteLen int + for shift := uint(0); ; shift += 7 { + if shift >= 64 { + return ErrIntOverflowQuery + } + if iNdEx >= l { + return io.ErrUnexpectedEOF + } + b := dAtA[iNdEx] + iNdEx++ + byteLen |= int(b&0x7F) << shift + if b < 0x80 { + break + } + } + if byteLen < 0 { + return ErrInvalidLengthQuery + } + postIndex := iNdEx + byteLen + if postIndex < 0 { + return ErrInvalidLengthQuery + } + if postIndex > l { + return io.ErrUnexpectedEOF + } + m.LastAppHash = append(m.LastAppHash[:0], dAtA[iNdEx:postIndex]...) + if m.LastAppHash == nil { + m.LastAppHash = []byte{} + } + iNdEx = postIndex + case 4: + if wireType != 2 { + return fmt.Errorf("proto: wrong wireType = %d for field BlockEvents", wireType) + } + var msglen int + for shift := uint(0); ; shift += 7 { + if shift >= 64 { + return ErrIntOverflowQuery + } + if iNdEx >= l { + return io.ErrUnexpectedEOF + } + b := dAtA[iNdEx] + iNdEx++ + msglen |= int(b&0x7F) << shift + if b < 0x80 { + break + } + } + if msglen < 0 { + return ErrInvalidLengthQuery + } + postIndex := iNdEx + msglen + if postIndex < 0 { + return ErrInvalidLengthQuery + } + if postIndex > l { + return io.ErrUnexpectedEOF + } + if err := m.BlockEvents.Unmarshal(dAtA[iNdEx:postIndex]); err != nil { + return err + } + iNdEx = postIndex + case 5: + if wireType != 2 { + return fmt.Errorf("proto: wrong wireType = %d for field TxEvents", wireType) + } + var msglen int + for shift := uint(0); ; shift += 7 { + if shift >= 64 { + return ErrIntOverflowQuery + } + if iNdEx >= l { + return io.ErrUnexpectedEOF + } + b := dAtA[iNdEx] + iNdEx++ + msglen |= int(b&0x7F) << shift + if b < 0x80 { + break + } + } + if msglen < 0 { + return ErrInvalidLengthQuery + } + postIndex := iNdEx + msglen + if postIndex < 0 { + return ErrInvalidLengthQuery + } + if postIndex > l { + return io.ErrUnexpectedEOF + } + m.TxEvents = append(m.TxEvents, EventSet{}) + if err := m.TxEvents[len(m.TxEvents)-1].Unmarshal(dAtA[iNdEx:postIndex]); err != nil { + return err + } + iNdEx = postIndex + case 6: + if wireType != 2 { + return fmt.Errorf("proto: wrong wireType = %d for field AdditionalDataEntries", wireType) + } + var msglen int + for shift := uint(0); ; shift += 7 { + if shift >= 64 { + return ErrIntOverflowQuery + } + if iNdEx >= l { + return io.ErrUnexpectedEOF + } + b := dAtA[iNdEx] + iNdEx++ + msglen |= int(b&0x7F) << shift + if b < 0x80 { + break + } + } + if msglen < 0 { + return ErrInvalidLengthQuery + } + postIndex := iNdEx + msglen + if postIndex < 0 { + return ErrInvalidLengthQuery + } + if postIndex > l { + return io.ErrUnexpectedEOF + } + m.AdditionalDataEntries = append(m.AdditionalDataEntries, &AdditionalDataEntry{}) + if err := m.AdditionalDataEntries[len(m.AdditionalDataEntries)-1].Unmarshal(dAtA[iNdEx:postIndex]); err != nil { + return err + } + iNdEx = postIndex + default: + iNdEx = preIndex + skippy, err := skipQuery(dAtA[iNdEx:]) + if err != nil { + return err + } + if (skippy < 0) || (iNdEx+skippy) < 0 { + return ErrInvalidLengthQuery + } + if (iNdEx + skippy) > l { + return io.ErrUnexpectedEOF + } + iNdEx += skippy + } + } + + if iNdEx > l { + return io.ErrUnexpectedEOF + } + return nil +} +func (m *AdditionalDataEntry) Unmarshal(dAtA []byte) error { + l := len(dAtA) + iNdEx := 0 + for iNdEx < l { + preIndex := iNdEx + var wire uint64 + for shift := uint(0); ; shift += 7 { + if shift >= 64 { + return ErrIntOverflowQuery + } + if iNdEx >= l { + return io.ErrUnexpectedEOF + } + b := dAtA[iNdEx] + iNdEx++ + wire |= uint64(b&0x7F) << shift + if b < 0x80 { + break + } + } + fieldNum := int32(wire >> 3) + wireType := int(wire & 0x7) + if wireType == 4 { + return fmt.Errorf("proto: AdditionalDataEntry: wiretype end group for non-group") + } + if fieldNum <= 0 { + return fmt.Errorf("proto: AdditionalDataEntry: illegal tag %d (wire type %d)", fieldNum, wire) + } + switch fieldNum { + case 1: + if wireType != 0 { + return fmt.Errorf("proto: wrong wireType = %d for field Type", wireType) + } + m.Type = 0 + for shift := uint(0); ; shift += 7 { + if shift >= 64 { + return ErrIntOverflowQuery + } + if iNdEx >= l { + return io.ErrUnexpectedEOF + } + b := dAtA[iNdEx] + iNdEx++ + m.Type |= AdditionalDataTypes(b&0x7F) << shift + if b < 0x80 { + break + } + } + case 2: + if wireType != 2 { + return fmt.Errorf("proto: wrong wireType = %d for field Data", wireType) + } + var byteLen int + for shift := uint(0); ; shift += 7 { + if shift >= 64 { + return ErrIntOverflowQuery + } + if iNdEx >= l { + return io.ErrUnexpectedEOF + } + b := dAtA[iNdEx] + iNdEx++ + byteLen |= int(b&0x7F) << shift + if b < 0x80 { + break + } + } + if byteLen < 0 { + return ErrInvalidLengthQuery + } + postIndex := iNdEx + byteLen + if postIndex < 0 { + return ErrInvalidLengthQuery + } + if postIndex > l { + return io.ErrUnexpectedEOF + } + m.Data = append(m.Data[:0], dAtA[iNdEx:postIndex]...) + if m.Data == nil { + m.Data = []byte{} + } + iNdEx = postIndex + default: + iNdEx = preIndex + skippy, err := skipQuery(dAtA[iNdEx:]) + if err != nil { + return err + } + if (skippy < 0) || (iNdEx+skippy) < 0 { + return ErrInvalidLengthQuery + } + if (iNdEx + skippy) > l { + return io.ErrUnexpectedEOF + } + iNdEx += skippy + } + } + + if iNdEx > l { + return io.ErrUnexpectedEOF + } + return nil +} +func (m *EventSet) Unmarshal(dAtA []byte) error { + l := len(dAtA) + iNdEx := 0 + for iNdEx < l { + preIndex := iNdEx + var wire uint64 + for shift := uint(0); ; shift += 7 { + if shift >= 64 { + return ErrIntOverflowQuery + } + if iNdEx >= l { + return io.ErrUnexpectedEOF + } + b := dAtA[iNdEx] + iNdEx++ + wire |= uint64(b&0x7F) << shift + if b < 0x80 { + break + } + } + fieldNum := int32(wire >> 3) + wireType := int(wire & 0x7) + if wireType == 4 { + return fmt.Errorf("proto: EventSet: wiretype end group for non-group") + } + if fieldNum <= 0 { + return fmt.Errorf("proto: EventSet: illegal tag %d (wire type %d)", fieldNum, wire) + } + switch fieldNum { + case 1: + if wireType != 2 { + return fmt.Errorf("proto: wrong wireType = %d for field PublishedEvents", wireType) + } + var byteLen int + for shift := uint(0); ; shift += 7 { + if shift >= 64 { + return ErrIntOverflowQuery + } + if iNdEx >= l { + return io.ErrUnexpectedEOF + } + b := dAtA[iNdEx] + iNdEx++ + byteLen |= int(b&0x7F) << shift + if b < 0x80 { + break + } + } + if byteLen < 0 { + return ErrInvalidLengthQuery + } + postIndex := iNdEx + byteLen + if postIndex < 0 { + return ErrInvalidLengthQuery + } + if postIndex > l { + return io.ErrUnexpectedEOF + } + m.PublishedEvents = append(m.PublishedEvents, make([]byte, postIndex-iNdEx)) + copy(m.PublishedEvents[len(m.PublishedEvents)-1], dAtA[iNdEx:postIndex]) + iNdEx = postIndex + case 2: + if wireType == 0 { + var v EventType + for shift := uint(0); ; shift += 7 { + if shift >= 64 { + return ErrIntOverflowQuery + } + if iNdEx >= l { + return io.ErrUnexpectedEOF + } + b := dAtA[iNdEx] + iNdEx++ + v |= EventType(b&0x7F) << shift + if b < 0x80 { + break + } + } + m.TrueOrders = append(m.TrueOrders, v) + } else if wireType == 2 { + var packedLen int + for shift := uint(0); ; shift += 7 { + if shift >= 64 { + return ErrIntOverflowQuery + } + if iNdEx >= l { + return io.ErrUnexpectedEOF + } + b := dAtA[iNdEx] + iNdEx++ + packedLen |= int(b&0x7F) << shift + if b < 0x80 { + break + } + } + if packedLen < 0 { + return ErrInvalidLengthQuery + } + postIndex := iNdEx + packedLen + if postIndex < 0 { + return ErrInvalidLengthQuery + } + if postIndex > l { + return io.ErrUnexpectedEOF + } + var elementCount int + if elementCount != 0 && len(m.TrueOrders) == 0 { + m.TrueOrders = make([]EventType, 0, elementCount) + } + for iNdEx < postIndex { + var v EventType + for shift := uint(0); ; shift += 7 { + if shift >= 64 { + return ErrIntOverflowQuery + } + if iNdEx >= l { + return io.ErrUnexpectedEOF + } + b := dAtA[iNdEx] + iNdEx++ + v |= EventType(b&0x7F) << shift + if b < 0x80 { + break + } + } + m.TrueOrders = append(m.TrueOrders, v) + } + } else { + return fmt.Errorf("proto: wrong wireType = %d for field TrueOrders", wireType) + } + case 3: + if wireType != 2 { + return fmt.Errorf("proto: wrong wireType = %d for field AbciEvents", wireType) + } + var msglen int + for shift := uint(0); ; shift += 7 { + if shift >= 64 { + return ErrIntOverflowQuery + } + if iNdEx >= l { + return io.ErrUnexpectedEOF + } + b := dAtA[iNdEx] + iNdEx++ + msglen |= int(b&0x7F) << shift + if b < 0x80 { + break + } + } + if msglen < 0 { + return ErrInvalidLengthQuery + } + postIndex := iNdEx + msglen + if postIndex < 0 { + return ErrInvalidLengthQuery + } + if postIndex > l { + return io.ErrUnexpectedEOF + } + m.AbciEvents = append(m.AbciEvents, v1.Event{}) + if err := m.AbciEvents[len(m.AbciEvents)-1].Unmarshal(dAtA[iNdEx:postIndex]); err != nil { + return err + } + iNdEx = postIndex + default: + iNdEx = preIndex + skippy, err := skipQuery(dAtA[iNdEx:]) + if err != nil { + return err + } + if (skippy < 0) || (iNdEx+skippy) < 0 { + return ErrInvalidLengthQuery + } + if (iNdEx + skippy) > l { + return io.ErrUnexpectedEOF + } + iNdEx += skippy + } + } + + if iNdEx > l { + return io.ErrUnexpectedEOF + } + return nil +} +func skipQuery(dAtA []byte) (n int, err error) { + l := len(dAtA) + iNdEx := 0 + depth := 0 + for iNdEx < l { + var wire uint64 + for shift := uint(0); ; shift += 7 { + if shift >= 64 { + return 0, ErrIntOverflowQuery + } + if iNdEx >= l { + return 0, io.ErrUnexpectedEOF + } + b := dAtA[iNdEx] + iNdEx++ + wire |= (uint64(b) & 0x7F) << shift + if b < 0x80 { + break + } + } + wireType := int(wire & 0x7) + switch wireType { + case 0: + for shift := uint(0); ; shift += 7 { + if shift >= 64 { + return 0, ErrIntOverflowQuery + } + if iNdEx >= l { + return 0, io.ErrUnexpectedEOF + } + iNdEx++ + if dAtA[iNdEx-1] < 0x80 { + break + } + } + case 1: + iNdEx += 8 + case 2: + var length int + for shift := uint(0); ; shift += 7 { + if shift >= 64 { + return 0, ErrIntOverflowQuery + } + if iNdEx >= l { + return 0, io.ErrUnexpectedEOF + } + b := dAtA[iNdEx] + iNdEx++ + length |= (int(b) & 0x7F) << shift + if b < 0x80 { + break + } + } + if length < 0 { + return 0, ErrInvalidLengthQuery + } + iNdEx += length + case 3: + depth++ + case 4: + if depth == 0 { + return 0, ErrUnexpectedEndOfGroupQuery + } + depth-- + case 5: + iNdEx += 4 + default: + return 0, fmt.Errorf("proto: illegal wireType %d", wireType) + } + if iNdEx < 0 { + return 0, ErrInvalidLengthQuery + } + if depth == 0 { + return iNdEx, nil + } + } + return 0, io.ErrUnexpectedEOF +} + +var ( + ErrInvalidLengthQuery = fmt.Errorf("proto: negative length found during unmarshaling") + ErrIntOverflowQuery = fmt.Errorf("proto: integer overflow") + ErrUnexpectedEndOfGroupQuery = fmt.Errorf("proto: unexpected end of group") +) diff --git a/client/chain/context.go b/client/chain/context.go index 7e3f0d968..a28a9ef2d 100644 --- a/client/chain/context.go +++ b/client/chain/context.go @@ -82,8 +82,24 @@ func NewInterfaceRegistry() types.InterfaceRegistry { // NewTxConfig initializes new Cosmos TxConfig with certain signModes enabled. func NewTxConfig(signModes []signingtypes.SignMode) client.TxConfig { - marshaler, _ := createInjectiveProtoCodec() - return tx.NewTxConfig(marshaler, signModes) + return newEncodingConfig(signModes).TxConfig +} + +// NewEncodingConfig initializes the full Injective encoding config used by sdk-go clients. +func NewEncodingConfig() EncodingConfig { + return newEncodingConfig([]signingtypes.SignMode{ + signingtypes.SignMode_SIGN_MODE_DIRECT, + }) +} + +func newEncodingConfig(signModes []signingtypes.SignMode) EncodingConfig { + marshaler, interfaceRegistry := createInjectiveProtoCodec() + + return EncodingConfig{ + InterfaceRegistry: interfaceRegistry, + Marshaler: marshaler, + TxConfig: tx.NewTxConfig(marshaler, signModes), + } } // NewClientContext creates a new Cosmos Client context, where chainID @@ -94,15 +110,7 @@ func NewClientContext( ) (client.Context, error) { clientCtx := client.Context{} - marshaler, interfaceRegistry := createInjectiveProtoCodec() - - encodingConfig := EncodingConfig{ - InterfaceRegistry: interfaceRegistry, - Marshaler: marshaler, - TxConfig: tx.NewTxConfig(marshaler, []signingtypes.SignMode{ - signingtypes.SignMode_SIGN_MODE_DIRECT, - }), - } + encodingConfig := NewEncodingConfig() var keyInfo keyring.Record diff --git a/client/mq/detector/config.go b/client/mq/detector/config.go new file mode 100644 index 000000000..6a3f42ae5 --- /dev/null +++ b/client/mq/detector/config.go @@ -0,0 +1,89 @@ +package main + +import ( + "errors" + "fmt" + "strings" + "time" + + "github.com/google/uuid" +) + +const ( + flagMQDetectorKafkaBrokers = "kafka-brokers" + flagMQDetectorConsumerID = "consumer-id" + flagMQDetectorRawTopic = "raw-topic" + flagMQDetectorLatestTopic = "latest-topic" + flagMQDetectorFullNodes = "full-nodes" + flagMQDetectorControlToken = "control-token" + flagMQDetectorRequestTimeout = "request-timeout" + flagMQDetectorMessageTimeout = "message-timeout" +) + +type mqDetectorConfig struct { + KafkaBrokers []string + ConsumerID string + ConsumerIDFile string + RawTopic string + LatestTopic string + FullNodes []string + ControlToken string + RequestTimeout time.Duration + MessageTimeout time.Duration +} + +func (cfg mqDetectorConfig) Validate() error { + if len(cfg.KafkaBrokers) == 0 { + return errors.New("invalid MQ detector config: no Kafka brokers specified") + } + + if strings.TrimSpace(cfg.ConsumerID) == "" { + return errors.New("invalid MQ detector config: consumer id cannot be empty") + } + + for i, broker := range cfg.KafkaBrokers { + if strings.TrimSpace(broker) == "" { + return fmt.Errorf("invalid MQ detector config: Kafka broker #%d is empty", i+1) + } + } + + if strings.TrimSpace(cfg.LatestTopic) == "" { + return errors.New("invalid MQ detector config: latest topic cannot be empty") + } + + if strings.TrimSpace(cfg.RawTopic) == "" { + return errors.New("invalid MQ detector config: raw topic cannot be empty") + } + + if cfg.RawTopic == cfg.LatestTopic { + return errors.New("invalid MQ detector config: raw topic and latest topic must be different") + } + + if len(cfg.FullNodes) == 0 { + return errors.New("invalid MQ detector config: no full nodes specified") + } + + for i, node := range cfg.FullNodes { + if strings.TrimSpace(node) == "" { + return fmt.Errorf("invalid MQ detector config: full node URL #%d is empty", i+1) + } + } + + if cfg.RequestTimeout <= 0 { + return errors.New("invalid MQ detector config: request timeout must be positive") + } + + if cfg.MessageTimeout <= 0 { + return errors.New("invalid MQ detector config: message timeout must be positive") + } + + return nil +} + +func resolveConsumerID(flagValue string) string { + if consumerID := strings.TrimSpace(flagValue); consumerID != "" { + return consumerID + } + + return uuid.NewString() +} diff --git a/client/mq/detector/config_test.go b/client/mq/detector/config_test.go new file mode 100644 index 000000000..7e209bf87 --- /dev/null +++ b/client/mq/detector/config_test.go @@ -0,0 +1,37 @@ +package main + +import ( + "testing" + "time" + + "github.com/google/uuid" +) + +func TestResolveConsumerIDGeneratesWhenEmpty(t *testing.T) { + consumerID := resolveConsumerID(" ") + + if _, err := uuid.Parse(consumerID); err != nil { + t.Fatalf("expected generated UUID consumer id, got %q: %v", consumerID, err) + } +} + +func TestValidateRequiresConsumerID(t *testing.T) { + cfg := mqDetectorConfig{ + KafkaBrokers: []string{"broker-1:9092"}, + ConsumerID: " ", + RawTopic: "raw", + LatestTopic: "latest", + FullNodes: []string{"http://node-1:9999"}, + RequestTimeout: time.Second, + MessageTimeout: time.Second, + } + + err := cfg.Validate() + if err == nil { + t.Fatal("expected Validate to reject empty consumer id") + } + + if err.Error() != "invalid MQ detector config: consumer id cannot be empty" { + t.Fatalf("unexpected error: %v", err) + } +} diff --git a/client/mq/detector/main.go b/client/mq/detector/main.go new file mode 100644 index 000000000..0b29ba288 --- /dev/null +++ b/client/mq/detector/main.go @@ -0,0 +1,280 @@ +package main + +import ( + "context" + "encoding/hex" + "errors" + "flag" + "fmt" + "io" + "net/http" + "os" + "os/signal" + "slices" + "strings" + "syscall" + "time" + + sdklog "cosmossdk.io/log" + "github.com/twmb/franz-go/pkg/kgo" + + "github.com/InjectiveLabs/sdk-go/chain/mq/types" +) + +func main() { + var ( + kafkaBrokers = flag.String(flagMQDetectorKafkaBrokers, "", "Comma-separated Kafka broker addresses") + consumerID = flag.String(flagMQDetectorConsumerID, "", "Kafka consumer id; generated when omitted") + rawTopic = flag.String(flagMQDetectorRawTopic, "", "Topic name for raw messages") + latestTopic = flag.String(flagMQDetectorLatestTopic, "", "Topic name for latest messages") + fullNodes = flag.String(flagMQDetectorFullNodes, "", "Comma-separated full node control plane URLs") + controlToken = flag.String(flagMQDetectorControlToken, "", "Bearer token for full node control plane requests") + requestTimeout = flag.Duration(flagMQDetectorRequestTimeout, 10*time.Second, "Timeout for block requests") + messageTimeout = flag.Duration(flagMQDetectorMessageTimeout, 30*time.Second, "Message waiting timeout duration") + ) + + flag.Parse() + + cfg := mqDetectorConfig{ + ConsumerID: resolveConsumerID(*consumerID), + RawTopic: *rawTopic, + LatestTopic: *latestTopic, + ControlToken: *controlToken, + RequestTimeout: *requestTimeout, + MessageTimeout: *messageTimeout, + } + + if *kafkaBrokers != "" { + cfg.KafkaBrokers = strings.Split(*kafkaBrokers, ",") + } + + if *fullNodes != "" { + cfg.FullNodes = strings.Split(*fullNodes, ",") + } + + if err := startMQDetector(context.Background(), cfg); err != nil && !errors.Is(err, context.Canceled) { + _, _ = fmt.Fprintln(os.Stderr, err) + os.Exit(1) + } +} + +func startMQDetector(ctx context.Context, cfg mqDetectorConfig) error { + if err := cfg.Validate(); err != nil { + return err + } + + logger := sdklog.NewLogger(os.Stderr) + opts := []kgo.Opt{ + kgo.SeedBrokers(cfg.KafkaBrokers...), + kgo.ConsumerGroup(cfg.ConsumerID), + kgo.ConsumeTopics(cfg.RawTopic, cfg.LatestTopic), + kgo.ConsumeResetOffset(kgo.NewOffset().AtStart()), + } + + client, err := kgo.NewClient(opts...) + if err != nil { + return err + } + + defer client.Close() + + signalCtx, stopSignals := signal.NotifyContext(ctx, os.Interrupt, syscall.SIGTERM) + defer stopSignals() + + ctx, cancel := context.WithCancelCause(signalCtx) + defer cancel(nil) + + var ( + rawTopicCh = make(chan *types.EventStreamResponse, 1) + latestTopicCh = make(chan *types.EventStreamResponse, 1) + ) + + poll := func(ctx context.Context, raw, latest chan<- *types.EventStreamResponse) error { + for { + fetches := client.PollFetches(ctx) + if err := fetches.Err(); err != nil { + if errors.Is(err, context.Canceled) || errors.Is(ctx.Err(), context.Canceled) { + return err + } + + err = fmt.Errorf("poll kafka fetches: %w", err) + logger.Error("error polling fetches, stopping consumer loop", "err", err.Error()) + return err + } + + iter := fetches.RecordIter() + for !iter.Done() { + record := iter.Next() + var msg types.EventStreamResponse + if err := msg.Unmarshal(record.Value); err != nil { + logger.Error("error decoding event message", "err", err.Error()) + continue + } + + switch record.Topic { + case cfg.LatestTopic: + latest <- &msg + case cfg.RawTopic: + raw <- &msg + default: + err := fmt.Errorf("unknown kafka topic %q", record.Topic) + logger.Error("unknown topic", "topic", record.Topic) + return err + } + } + } + } + + controlPlaneClient := &http.Client{Timeout: cfg.RequestTimeout} + callControlPlane := func(startHeight uint64) { + logger.Info("calling control plane", "startHeight", startHeight) + for _, node := range cfg.FullNodes { + if err := requestControlPlaneBlocks(ctx, controlPlaneClient, startHeight, node, cfg.ControlToken); err != nil { + logger.Error("error requesting blocks from node", "node", node, "err", err.Error()) + } + + logger.Debug("requested blocks from node", "node", node) + } + } + + // start the kafka polling loop + go func() { + defer func() { + close(rawTopicCh) + close(latestTopicCh) + }() + + if err := poll(ctx, rawTopicCh, latestTopicCh); err != nil { + if !errors.Is(err, context.Canceled) && !errors.Is(ctx.Err(), context.Canceled) { + cancel(err) + return + } + } + + cancel(nil) + }() + + var ( + latestHeight = int64(0) + rawHeights = make(map[int64]struct{}) + lastSeenLatest time.Time + ) + + for { + if noLatestForAWhile := !lastSeenLatest.IsZero() && time.Since(lastSeenLatest) > cfg.MessageTimeout; noLatestForAWhile { + go callControlPlane(uint64(latestHeight) + 1) + } + + select { + case <-ctx.Done(): + if err := context.Cause(ctx); err != nil { + return err + } + + return ctx.Err() + case msg, ok := <-rawTopicCh: + if msg == nil || !ok { + if err := context.Cause(ctx); err != nil && !errors.Is(err, context.Canceled) { + return err + } + + return nil + } + + if msg.BlockHeight <= latestHeight { + continue // no need to store previous heights + } + + if _, ok := rawHeights[msg.BlockHeight]; ok { + continue // already here + } + + rawHeights[msg.BlockHeight] = struct{}{} + logger.Info("received raw event", + "height", msg.BlockHeight, + "app_hash", hex.EncodeToString(msg.AppHash), + "last_app_hash", hex.EncodeToString(msg.LastAppHash), + ) + case msg, ok := <-latestTopicCh: + if msg == nil || !ok { + if err := context.Cause(ctx); err != nil && !errors.Is(err, context.Canceled) { + return err + } + + return nil + } + + latestHeight = msg.BlockHeight + lastSeenLatest = time.Now() + logger.Info("received latest event", + "height", msg.BlockHeight, + "app_hash", hex.EncodeToString(msg.AppHash), + "last_app_hash", hex.EncodeToString(msg.LastAppHash), + ) + + for h := range rawHeights { + if h <= latestHeight { + delete(rawHeights, h) + } + } + } + + if nothingYet := latestHeight == 0 && len(rawHeights) == 0; nothingYet { + continue + } + + if needLatestHeightFirst := latestHeight == 0 && len(rawHeights) != 0; needLatestHeightFirst { + continue + } + + if needSomeRawHeightAlso := latestHeight != 0 && len(rawHeights) == 0; needSomeRawHeightAlso { + continue + } + + sortedHeights := make([]int64, 0, len(rawHeights)) + for h := range rawHeights { + sortedHeights = append(sortedHeights, h) + } + + slices.Sort(sortedHeights) + + if thereIsAGap := latestHeight+1 < sortedHeights[0]; thereIsAGap { + logger.Warn("height gap detected", "want", latestHeight+1, "got", sortedHeights[0]) + go callControlPlane(uint64(latestHeight) + 1) + } + } +} + +func requestControlPlaneBlocks( + ctx context.Context, + client *http.Client, + startHeight uint64, + nodeURL, + controlToken string, +) error { + requestURL := fmt.Sprintf("%s/request?from_height=%d", nodeURL, startHeight) + req, err := http.NewRequestWithContext(ctx, http.MethodPost, requestURL, http.NoBody) + if err != nil { + return err + } + + if controlToken != "" { + req.Header.Set("Authorization", "Bearer "+controlToken) + } + + resp, err := client.Do(req) + if err != nil { + return err + } + + defer func() { + _ = resp.Body.Close() + }() + + if resp.StatusCode != http.StatusAccepted && resp.StatusCode != http.StatusOK { + body, _ := io.ReadAll(resp.Body) + return fmt.Errorf("unexpected status code: %d, body: %s", resp.StatusCode, string(body)) + } + + return nil +} diff --git a/client/mq/detector/main_test.go b/client/mq/detector/main_test.go new file mode 100644 index 000000000..401c672ee --- /dev/null +++ b/client/mq/detector/main_test.go @@ -0,0 +1,52 @@ +package main + +import ( + "context" + "net/http" + "net/http/httptest" + "testing" +) + +func TestRequestControlPlaneBlocksAddsBearerToken(t *testing.T) { + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if r.Method != http.MethodPost { + t.Fatalf("unexpected method: %s", r.Method) + } + + if r.URL.Path != "/request" { + t.Fatalf("unexpected path: %s", r.URL.Path) + } + + if got := r.URL.Query().Get("from_height"); got != "42" { + t.Fatalf("unexpected from_height: %q", got) + } + + if got := r.Header.Get("Authorization"); got != "Bearer secret" { + t.Fatalf("unexpected Authorization header: %q", got) + } + + w.WriteHeader(http.StatusAccepted) + })) + defer server.Close() + + err := requestControlPlaneBlocks(context.Background(), server.Client(), 42, server.URL, "secret") + if err != nil { + t.Fatalf("requestControlPlaneBlocks returned error: %v", err) + } +} + +func TestRequestControlPlaneBlocksOmitsBearerTokenWhenEmpty(t *testing.T) { + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if got := r.Header.Get("Authorization"); got != "" { + t.Fatalf("unexpected Authorization header: %q", got) + } + + w.WriteHeader(http.StatusOK) + })) + defer server.Close() + + err := requestControlPlaneBlocks(context.Background(), server.Client(), 42, server.URL, "") + if err != nil { + t.Fatalf("requestControlPlaneBlocks returned error: %v", err) + } +} diff --git a/client/mq/stream/client/config.go b/client/mq/stream/client/config.go new file mode 100644 index 000000000..883d24819 --- /dev/null +++ b/client/mq/stream/client/config.go @@ -0,0 +1,99 @@ +package main + +import ( + "errors" + "fmt" + "os" + "path/filepath" + "strings" + + "github.com/google/uuid" +) + +const ( + flagMQStreamClientAddress = "address" + flagMQStreamClientConsumerID = "consumer-id" + flagMQStreamClientConsumerIDFile = "consumer-id-file" + flagMQStreamClientTopic = "topic" + flagMQStreamClientFormat = "format" + flagMQStreamClientEventsDir = "events-dir" + + defaultConsumerIDFile = ".mq-stream-consumer-id" +) + +type mqStreamClientConfig struct { + Address string + ConsumerID string + ConsumerIDFile string + Topic string + Format string + EventsDir string +} + +func (cfg mqStreamClientConfig) Validate() error { + if strings.TrimSpace(cfg.Address) == "" { + return errors.New("invalid MQ stream client config: address cannot be empty") + } + + if strings.TrimSpace(cfg.ConsumerID) == "" { + return errors.New("invalid MQ stream client config: consumer id cannot be empty") + } + + if strings.TrimSpace(cfg.Topic) == "" { + return errors.New("invalid MQ stream client config: topic cannot be empty") + } + + switch cfg.Format { + case "minimal", "verbose": + return nil + default: + return fmt.Errorf("invalid MQ stream client config: unsupported format %q", cfg.Format) + } +} + +func resolveConsumerID(flagValue, filePath string) (string, error) { + if consumerID := strings.TrimSpace(flagValue); consumerID != "" { + return consumerID, nil + } + + filePath = strings.TrimSpace(filePath) + if filePath == "" { + return uuid.NewString(), nil + } + + bz, err := os.ReadFile(filePath) + if err == nil { + consumerID := strings.TrimSpace(string(bz)) + if consumerID == "" { + return "", fmt.Errorf("consumer id file %q is empty", filePath) + } + + return consumerID, nil + } + + if !errors.Is(err, os.ErrNotExist) { + return "", fmt.Errorf("read consumer id file: %w", err) + } + + consumerID := uuid.NewString() + if err := writeConsumerIDFile(filePath, consumerID); err != nil { + return "", err + } + + return consumerID, nil +} + +func writeConsumerIDFile(filePath, consumerID string) error { + dir := filepath.Dir(filePath) + if dir != "." { + if err := os.MkdirAll(dir, 0o750); err != nil { + return fmt.Errorf("create consumer id file dir: %w", err) + } + } + + if err := os.WriteFile(filePath, []byte(consumerID+"\n"), 0o600); err != nil { + return fmt.Errorf("write consumer id file: %w", err) + } + + return nil +} diff --git a/client/mq/stream/client/main.go b/client/mq/stream/client/main.go new file mode 100644 index 000000000..ff13fa8b9 --- /dev/null +++ b/client/mq/stream/client/main.go @@ -0,0 +1,137 @@ +package main + +import ( + "context" + "errors" + "flag" + "fmt" + "io" + "os" + "strings" + "time" + + "google.golang.org/grpc" + "google.golang.org/grpc/codes" + "google.golang.org/grpc/credentials/insecure" + "google.golang.org/grpc/keepalive" + "google.golang.org/grpc/status" + + mqtypes "github.com/InjectiveLabs/sdk-go/chain/mq/types" +) + +var kacp = keepalive.ClientParameters{ + Time: 30 * time.Second, // send pings every 30 seconds if there is no activity + Timeout: 5 * time.Second, // wait 5 second for ping ack before considering the connection dead + PermitWithoutStream: false, // do not send pings without active streams +} + +func main() { + var ( + address = flag.String(flagMQStreamClientAddress, "localhost:9988", "MQ gRPC stream server address") + consumerID = flag.String(flagMQStreamClientConsumerID, "", "Stable Kafka consumer id") + consumerIDFile = flag.String(flagMQStreamClientConsumerIDFile, defaultConsumerIDFile, "File used to persist a generated Kafka consumer id") //nolint // it's fine + topic = flag.String(flagMQStreamClientTopic, "", "Topic to consume") + format = flag.String(flagMQStreamClientFormat, "verbose", "Output format: verbose or minimal") + eventsDir = flag.String(flagMQStreamClientEventsDir, "", "Directory to write one JSON file per streamed block") + ) + + flag.Parse() + + cfg := mqStreamClientConfig{ + Address: *address, + ConsumerID: *consumerID, + ConsumerIDFile: *consumerIDFile, + Topic: *topic, + Format: *format, + EventsDir: *eventsDir, + } + + if err := startMQStreamClient(context.Background(), cfg); err != nil && !errors.Is(err, context.Canceled) { + _, _ = fmt.Fprintln(os.Stderr, err) + os.Exit(1) + } +} + +func startMQStreamClient(ctx context.Context, cfg mqStreamClientConfig) error { + if err := cfg.Validate(); err != nil { + return err + } + + resolvedConsumerID, err := resolveConsumerID(cfg.ConsumerID, cfg.ConsumerIDFile) + if err != nil { + return err + } + + cdc, err := newPublishEventDecoder() + if err != nil { + return fmt.Errorf("failed to initialize publish event decoder: %w", err) + } + + cc, err := grpc.NewClient( + cfg.Address, + grpc.WithTransportCredentials(insecure.NewCredentials()), + grpc.WithKeepaliveParams(kacp), + ) + if err != nil { + return fmt.Errorf("failed to connect: %w", err) + } + + defer func() { + _ = cc.Close() + }() + + stream, err := mqtypes.NewEventStreamClient(cc).EventStream(ctx, &mqtypes.EventStreamRequest{ + ConsumerId: resolvedConsumerID, + Topic: cfg.Topic, + }) + if err != nil { + return fmt.Errorf("failed to start event stream: %w", err) + } + + for { + res, err := stream.Recv() + if err != nil { + if isExpectedStreamClose(err) { + return nil + } + + return fmt.Errorf("event stream failed: %w", err) + } + + if err := writeEventsFile(cfg.EventsDir, res, cdc); err != nil { + return fmt.Errorf("failed to write events file: %w", err) + } + + switch cfg.Format { + case "minimal": + printMinimal(res) + case "verbose": + printVerbose(res, cdc) + default: + return fmt.Errorf("unsupported format %q", cfg.Format) + } + } +} + +func isExpectedStreamClose(err error) bool { + if errors.Is(err, context.Canceled) || errors.Is(err, io.EOF) { + return true + } + + st, ok := status.FromError(err) + if !ok { + return false + } + + switch st.Code() { + case codes.Canceled: + return true + case codes.Unavailable: + msg := st.Message() + return strings.Contains(msg, "EOF") || + strings.Contains(msg, "transport is closing") || + strings.Contains(msg, "client connection is closing") + default: + return false + } +} diff --git a/client/mq/stream/client/util.go b/client/mq/stream/client/util.go new file mode 100644 index 000000000..846232c49 --- /dev/null +++ b/client/mq/stream/client/util.go @@ -0,0 +1,247 @@ +package main + +import ( + "encoding/base64" + "encoding/hex" + "encoding/json" + "errors" + "fmt" + "os" + "path/filepath" + "time" + "unicode/utf8" + + abcitypes "github.com/cometbft/cometbft/api/cometbft/abci/v1" + sdkcodec "github.com/cosmos/cosmos-sdk/codec" + codectypes "github.com/cosmos/cosmos-sdk/codec/types" + + mqtypes "github.com/InjectiveLabs/sdk-go/chain/mq/types" + chainclient "github.com/InjectiveLabs/sdk-go/client/chain" +) + +type blockEventsFile struct { + CapturedAt string `json:"captured_at"` + BlockHeight int64 `json:"block_height"` + AppHash string `json:"app_hash"` + LastAppHash string `json:"last_app_hash"` + BlockEvents eventSetEventsFile `json:"block_events"` + TxEvents []eventSetEventsFile `json:"tx_events"` + AdditionalDataEntries []additionalDataFile `json:"additional_data_entries,omitempty"` +} + +type eventSetEventsFile struct { + TrueOrders []string `json:"true_orders"` + ABCIEvents []abcitypes.Event `json:"abci_events"` + PublishEvents []publishEventFile `json:"publish_events"` +} + +type publishEventFile struct { + TypeURL string `json:"type_url"` + Value json.RawMessage `json:"value"` +} + +type encodedBytesFile struct { + Base64 string `json:"base64"` + UTF8 string `json:"utf8,omitempty"` +} + +type additionalDataFile struct { + Type string `json:"type"` + Base64 string `json:"base64"` + UTF8 string `json:"utf8,omitempty"` +} + +type publishEventDecoder struct { + cdc *sdkcodec.ProtoCodec +} + +func newPublishEventDecoder() (*publishEventDecoder, error) { + encodingConfig := chainclient.NewEncodingConfig() + cdc, ok := encodingConfig.Marshaler.(*sdkcodec.ProtoCodec) + if !ok { + return nil, errors.New("encoding config marshaler is not a proto codec") + } + + return &publishEventDecoder{cdc: cdc}, nil +} + +func shortHash(b []byte) string { + if len(b) == 0 { + return "-" + } + if len(b) > 8 { + b = b[:8] + } + return hex.EncodeToString(b) +} + +func encodeBytes(b []byte) encodedBytesFile { + encoded := encodedBytesFile{Base64: base64.StdEncoding.EncodeToString(b)} + if utf8.Valid(b) { + encoded.UTF8 = string(b) + } + return encoded +} + +func transformEventSet(events mqtypes.EventSet, decoder *publishEventDecoder) eventSetEventsFile { + trueOrders := make([]string, 0, len(events.TrueOrders)) + for _, order := range events.TrueOrders { + trueOrders = append(trueOrders, order.String()) + } + + publishEvents := make([]publishEventFile, 0, len(events.PublishedEvents)) + for idx, event := range events.PublishedEvents { + publishEvent, err := decoder.decodePublishEvent(event) + if err != nil { + _, _ = fmt.Fprintf(os.Stderr, "error decoding publish event %d: %v\n", idx, err) + continue + } + publishEvents = append(publishEvents, publishEvent) + } + + return eventSetEventsFile{ + TrueOrders: trueOrders, + ABCIEvents: events.AbciEvents, + PublishEvents: publishEvents, + } +} + +func transformAdditionalData(entries []*mqtypes.AdditionalDataEntry) []additionalDataFile { + result := make([]additionalDataFile, 0, len(entries)) + for _, entry := range entries { + data := encodeBytes(entry.Data) + result = append(result, additionalDataFile{ + Type: entry.Type.String(), + Base64: data.Base64, + UTF8: data.UTF8, + }) + } + + return result +} + +func writeEventsFile(eventsDir string, res *mqtypes.EventStreamResponse, decoder *publishEventDecoder) error { + if eventsDir == "" { + return nil + } + + txEvents := make([]eventSetEventsFile, 0, len(res.TxEvents)) + for _, events := range res.TxEvents { + txEventSet := transformEventSet(events, decoder) + txEvents = append(txEvents, txEventSet) + } + + blockEvents := transformEventSet(res.BlockEvents, decoder) + + out := blockEventsFile{ + CapturedAt: time.Now().Format(time.RFC3339), + BlockHeight: res.BlockHeight, + AppHash: hex.EncodeToString(res.AppHash), + LastAppHash: hex.EncodeToString(res.LastAppHash), + BlockEvents: blockEvents, + TxEvents: txEvents, + AdditionalDataEntries: transformAdditionalData(res.AdditionalDataEntries), + } + + if err := os.MkdirAll(eventsDir, 0o750); err != nil { + return fmt.Errorf("create events dir: %w", err) + } + + bz, err := json.MarshalIndent(out, "", " ") + if err != nil { + return fmt.Errorf("marshal events file: %w", err) + } + + filename := filepath.Join(eventsDir, fmt.Sprintf("block-%012d.json", res.BlockHeight)) + if err := os.WriteFile(filename, append(bz, '\n'), 0o600); err != nil { + return fmt.Errorf("write events file: %w", err) + } + + return nil +} + +func printMinimal(res *mqtypes.EventStreamResponse) { + _, _ = fmt.Printf( + "%s height=%d tx_events=%d abci_events=%d publish_events=%d app_hash=%s last_app_hash=%s\n", + time.Now().Format(time.RFC3339), + res.BlockHeight, + len(res.TxEvents), + len(res.BlockEvents.AbciEvents), + len(res.BlockEvents.PublishedEvents), + shortHash(res.AppHash), + shortHash(res.LastAppHash), + ) +} + +func printVerbose(res *mqtypes.EventStreamResponse, decoder *publishEventDecoder) { + _, _ = fmt.Println("---------------------") + _, _ = fmt.Println("height : ", res.BlockHeight) + _, _ = fmt.Println("app hash : ", hex.EncodeToString(res.AppHash)) + _, _ = fmt.Println("last app hash : ", hex.EncodeToString(res.LastAppHash)) + + _, _ = fmt.Println("block publish events : ") + for _, e := range res.BlockEvents.PublishedEvents { + printPublishEvent(e, decoder) + } + + _, _ = fmt.Println("block events true order : ", res.BlockEvents.TrueOrders) + + for i, tx := range res.TxEvents { + _, _ = fmt.Println("publish events for tx index : ", i) + for _, e := range tx.PublishedEvents { + printPublishEvent(e, decoder) + } + _, _ = fmt.Println("tx events true order : ", tx.TrueOrders) + } +} + +func printPublishEvent(event []byte, decoder *publishEventDecoder) { + publishEvent, err := decoder.decodePublishEvent(event) + if err != nil { + _, _ = fmt.Printf("error decoding publish event: %v\n", err) + return + } + + bz, err := json.MarshalIndent(publishEvent, "", " ") + if err != nil { + _, _ = fmt.Printf("error formatting publish event: %v\n", err) + return + } + _, _ = fmt.Println(string(bz)) +} + +func (d *publishEventDecoder) decodePublishEvent(event []byte) (publishEventFile, error) { + if d == nil || d.cdc == nil { + return publishEventFile{}, errors.New("publish event decoder is not initialized") + } + + var anyEvent codectypes.Any + if err := d.cdc.Unmarshal(event, &anyEvent); err != nil { + return publishEventFile{}, fmt.Errorf("decode publish event any: %w", err) + } + + if anyEvent.TypeUrl == "" { + return publishEventFile{}, errors.New("decode publish event any: missing type_url") + } + + anyJSON, err := sdkcodec.ProtoMarshalJSON(&anyEvent, nil) + if err != nil { + return publishEventFile{}, fmt.Errorf("marshal publish event any json %q: %w", anyEvent.TypeUrl, err) + } + + var value map[string]json.RawMessage + if err := json.Unmarshal(anyJSON, &value); err != nil { + return publishEventFile{}, fmt.Errorf("decode publish event json %q: %w", anyEvent.TypeUrl, err) + } + delete(value, "@type") + + valueJSON, err := json.Marshal(value) + if err != nil { + return publishEventFile{}, fmt.Errorf("marshal publish event json %q: %w", anyEvent.TypeUrl, err) + } + + return publishEventFile{ + TypeURL: anyEvent.TypeUrl, + Value: json.RawMessage(valueJSON), + }, nil +} diff --git a/client/mq/stream/client/util_test.go b/client/mq/stream/client/util_test.go new file mode 100644 index 000000000..55f499240 --- /dev/null +++ b/client/mq/stream/client/util_test.go @@ -0,0 +1,456 @@ +package main + +import ( + "context" + "encoding/json" + "errors" + "fmt" + "io" + "os" + "path/filepath" + "testing" + "time" + + sdkmath "cosmossdk.io/math" + abcitypes "github.com/cometbft/cometbft/api/cometbft/abci/v1" + codectypes "github.com/cosmos/cosmos-sdk/codec/types" + sdk "github.com/cosmos/cosmos-sdk/types" + "github.com/cosmos/gogoproto/proto" + "github.com/google/uuid" + "github.com/stretchr/testify/require" + "google.golang.org/grpc/codes" + "google.golang.org/grpc/status" + + exchangetypes "github.com/InjectiveLabs/sdk-go/chain/exchange/types" + mqtypes "github.com/InjectiveLabs/sdk-go/chain/mq/types" +) + +func TestDecodePublishEventRendersAnyAsJSON(t *testing.T) { + t.Parallel() + + event := &exchangetypes.EventSubaccountWithdraw{ + Amount: sdk.NewCoin("inj", sdkmath.NewInt(123123)), + } + eventBz := packPublishEventForClientTest(t, event) + decoder := newTestPublishEventDecoder(t) + + decoded, err := decoder.decodePublishEvent(eventBz) + + require.NoError(t, err) + require.Equal(t, codectypes.MsgTypeURL(event), decoded.TypeURL) + + var value map[string]json.RawMessage + require.NoError(t, json.Unmarshal(decoded.Value, &value)) + require.JSONEq(t, `{"denom":"inj","amount":"123123"}`, string(value["amount"])) +} + +func TestShortHash(t *testing.T) { + t.Parallel() + + testCases := []struct { + name string + input []byte + expected string + }{ + { + name: "empty", + expected: "-", + }, + { + name: "short", + input: []byte{0x01, 0x02, 0x0f}, + expected: "01020f", + }, + { + name: "truncated", + input: []byte{0x01, 0x02, 0x03, 0x04, 0x05, 0x06, 0x07, 0x08, 0x09}, + expected: "0102030405060708", + }, + } + + for _, tc := range testCases { + t.Run(tc.name, func(t *testing.T) { + require.Equal(t, tc.expected, shortHash(tc.input)) + }) + } +} + +func TestEncodeBytesIncludesUTF8OnlyForValidData(t *testing.T) { + t.Parallel() + + require.Equal(t, encodedBytesFile{ + Base64: "aGVsbG8=", + UTF8: "hello", + }, encodeBytes([]byte("hello"))) + require.Equal(t, encodedBytesFile{ + Base64: "//4=", + }, encodeBytes([]byte{0xff, 0xfe})) +} + +func TestTransformEventSet(t *testing.T) { + t.Parallel() + + event := &exchangetypes.EventSubaccountWithdraw{ + Amount: sdk.NewCoin("inj", sdkmath.NewInt(123123)), + } + abciEvent := abcitypes.Event{ + Type: "block.event", + Attributes: []abcitypes.EventAttribute{ + {Key: "scope", Value: "block", Index: true}, + }, + } + decoder := newTestPublishEventDecoder(t) + + transformed := transformEventSet(mqtypes.EventSet{ + PublishedEvents: [][]byte{packPublishEventForClientTest(t, event)}, + TrueOrders: []mqtypes.EventType{ + mqtypes.EventType_ABCI, + mqtypes.EventType_PUBLISH, + }, + AbciEvents: []abcitypes.Event{abciEvent}, + }, decoder) + + require.Equal(t, []string{"ABCI", "PUBLISH"}, transformed.TrueOrders) + require.Equal(t, []abcitypes.Event{abciEvent}, transformed.ABCIEvents) + require.Len(t, transformed.PublishEvents, 1) + require.Equal(t, codectypes.MsgTypeURL(event), transformed.PublishEvents[0].TypeURL) + + var value map[string]json.RawMessage + require.NoError(t, json.Unmarshal(transformed.PublishEvents[0].Value, &value)) + require.JSONEq(t, `{"denom":"inj","amount":"123123"}`, string(value["amount"])) +} + +func TestTransformEventSetSkipsMalformedPublishEvent(t *testing.T) { + t.Parallel() + + decoder := newTestPublishEventDecoder(t) + + transformed := transformEventSet(mqtypes.EventSet{ + PublishedEvents: [][]byte{{0xff}}, + }, decoder) + + require.Empty(t, transformed.PublishEvents) +} + +func TestTransformAdditionalData(t *testing.T) { + t.Parallel() + + require.Equal(t, []additionalDataFile{ + { + Type: "EVM", + Base64: "ZXZtLWRhdGE=", + UTF8: "evm-data", + }, + { + Type: "EVM", + Base64: "//4=", + }, + }, transformAdditionalData([]*mqtypes.AdditionalDataEntry{ + { + Type: mqtypes.AdditionalDataTypes_EVM, + Data: []byte("evm-data"), + }, + { + Type: mqtypes.AdditionalDataTypes_EVM, + Data: []byte{0xff, 0xfe}, + }, + })) +} + +func TestDecodePublishEventRejectsRawBytes(t *testing.T) { + t.Parallel() + + event := &exchangetypes.EventSubaccountWithdraw{ + Amount: sdk.NewCoin("inj", sdkmath.NewInt(123123)), + } + eventBz, err := proto.Marshal(event) + require.NoError(t, err) + + decoder := newTestPublishEventDecoder(t) + _, err = decoder.decodePublishEvent(eventBz) + + require.ErrorContains(t, err, "missing type_url") +} + +func TestDecodePublishEventRejectsUnknownTypeURL(t *testing.T) { + t.Parallel() + + eventBz, err := proto.Marshal(&codectypes.Any{ + TypeUrl: "/injective.unknown.Event", + Value: []byte{0x08, 0x01}, + }) + require.NoError(t, err) + + decoder := newTestPublishEventDecoder(t) + _, err = decoder.decodePublishEvent(eventBz) + + require.ErrorContains(t, err, `marshal publish event any json "/injective.unknown.Event"`) +} + +func TestDecodePublishEventRejectsMalformedAnyValue(t *testing.T) { + t.Parallel() + + event := &exchangetypes.EventSubaccountWithdraw{} + eventBz, err := proto.Marshal(&codectypes.Any{ + TypeUrl: codectypes.MsgTypeURL(event), + Value: []byte{0xff}, + }) + require.NoError(t, err) + + decoder := newTestPublishEventDecoder(t) + _, err = decoder.decodePublishEvent(eventBz) + + require.ErrorContains(t, err, "marshal publish event any json") + require.ErrorContains(t, err, codectypes.MsgTypeURL(event)) +} + +func TestWriteEventsFileCreatesDirectoryWithRestrictedPermissions(t *testing.T) { + t.Parallel() + + eventsDir := filepath.Join(t.TempDir(), "events") + decoder := newTestPublishEventDecoder(t) + + err := writeEventsFile(eventsDir, &mqtypes.EventStreamResponse{BlockHeight: 7}, decoder) + require.NoError(t, err) + + info, err := os.Stat(eventsDir) + require.NoError(t, err) + require.True(t, info.IsDir()) + require.Zero(t, info.Mode().Perm()&^os.FileMode(0o750)) + + fileInfo, err := os.Stat(filepath.Join(eventsDir, "block-000000000007.json")) + require.NoError(t, err) + require.Zero(t, fileInfo.Mode().Perm()&^os.FileMode(0o600)) +} + +func TestWriteEventsFileSkipsWhenDirectoryUnset(t *testing.T) { + t.Parallel() + + err := writeEventsFile("", &mqtypes.EventStreamResponse{ + BlockEvents: mqtypes.EventSet{ + PublishedEvents: [][]byte{{0xff}}, + }, + }, nil) + + require.NoError(t, err) +} + +func TestResolveConsumerIDUsesFlagValue(t *testing.T) { + t.Parallel() + + filePath := filepath.Join(t.TempDir(), "consumer-id") + + consumerID, err := resolveConsumerID(" from-flag ", filePath) + + require.NoError(t, err) + require.Equal(t, "from-flag", consumerID) + require.NoFileExists(t, filePath) +} + +func TestResolveConsumerIDReadsExistingFile(t *testing.T) { + t.Parallel() + + filePath := filepath.Join(t.TempDir(), "consumer-id") + require.NoError(t, os.WriteFile(filePath, []byte("from-file\n"), 0o600)) + + consumerID, err := resolveConsumerID("", filePath) + + require.NoError(t, err) + require.Equal(t, "from-file", consumerID) +} + +func TestResolveConsumerIDGeneratesAndWritesMissingFile(t *testing.T) { + t.Parallel() + + filePath := filepath.Join(t.TempDir(), "nested", "consumer-id") + + consumerID, err := resolveConsumerID("", filePath) + + require.NoError(t, err) + require.NotEmpty(t, consumerID) + _, err = uuid.Parse(consumerID) + require.NoError(t, err) + + bz, err := os.ReadFile(filePath) + require.NoError(t, err) + require.Equal(t, consumerID+"\n", string(bz)) +} + +func TestWriteEventsFileWritesStructuredPayload(t *testing.T) { + t.Parallel() + + event := &exchangetypes.EventSubaccountWithdraw{ + Amount: sdk.NewCoin("inj", sdkmath.NewInt(123123)), + } + blockABCIEvent := abcitypes.Event{ + Type: "block.event", + Attributes: []abcitypes.EventAttribute{ + {Key: "scope", Value: "block", Index: true}, + }, + } + eventsDir := t.TempDir() + decoder := newTestPublishEventDecoder(t) + + err := writeEventsFile(eventsDir, &mqtypes.EventStreamResponse{ + BlockHeight: 11, + AppHash: []byte{0x01, 0x02, 0x03}, + LastAppHash: []byte{0x04, 0x05, 0x06}, + BlockEvents: mqtypes.EventSet{ + TrueOrders: []mqtypes.EventType{mqtypes.EventType_ABCI}, + AbciEvents: []abcitypes.Event{blockABCIEvent}, + }, + TxEvents: []mqtypes.EventSet{ + { + PublishedEvents: [][]byte{packPublishEventForClientTest(t, event)}, + TrueOrders: []mqtypes.EventType{mqtypes.EventType_PUBLISH}, + }, + }, + AdditionalDataEntries: []*mqtypes.AdditionalDataEntry{ + { + Type: mqtypes.AdditionalDataTypes_EVM, + Data: []byte("extra"), + }, + }, + }, decoder) + + require.NoError(t, err) + + bz, err := os.ReadFile(filepath.Join(eventsDir, "block-000000000011.json")) + require.NoError(t, err) + + var written blockEventsFile + require.NoError(t, json.Unmarshal(bz, &written)) + require.Equal(t, int64(11), written.BlockHeight) + require.Equal(t, "010203", written.AppHash) + require.Equal(t, "040506", written.LastAppHash) + require.Equal(t, []string{"ABCI"}, written.BlockEvents.TrueOrders) + require.Equal(t, []abcitypes.Event{blockABCIEvent}, written.BlockEvents.ABCIEvents) + require.Len(t, written.TxEvents, 1) + require.Equal(t, []string{"PUBLISH"}, written.TxEvents[0].TrueOrders) + require.Len(t, written.TxEvents[0].PublishEvents, 1) + require.Equal(t, codectypes.MsgTypeURL(event), written.TxEvents[0].PublishEvents[0].TypeURL) + require.Equal(t, []additionalDataFile{ + { + Type: "EVM", + Base64: "ZXh0cmE=", + UTF8: "extra", + }, + }, written.AdditionalDataEntries) + _, err = time.Parse(time.RFC3339, written.CapturedAt) + require.NoError(t, err) +} + +func TestWriteEventsFileSkipsMalformedPublishEvents(t *testing.T) { + t.Parallel() + + eventsDir := t.TempDir() + decoder := newTestPublishEventDecoder(t) + + err := writeEventsFile(eventsDir, &mqtypes.EventStreamResponse{ + BlockHeight: 1, + BlockEvents: mqtypes.EventSet{ + PublishedEvents: [][]byte{{0xff}}, + }, + }, decoder) + require.NoError(t, err) + + err = writeEventsFile(eventsDir, &mqtypes.EventStreamResponse{ + BlockHeight: 2, + TxEvents: []mqtypes.EventSet{ + { + PublishedEvents: [][]byte{{0xff}}, + }, + }, + }, decoder) + require.NoError(t, err) + + for _, height := range []int{1, 2} { + bz, err := os.ReadFile(filepath.Join(eventsDir, fmt.Sprintf("block-%012d.json", height))) + require.NoError(t, err) + + var written blockEventsFile + require.NoError(t, json.Unmarshal(bz, &written)) + require.Empty(t, written.BlockEvents.PublishEvents) + for _, txEvents := range written.TxEvents { + require.Empty(t, txEvents.PublishEvents) + } + } +} + +func TestIsExpectedStreamClose(t *testing.T) { + t.Parallel() + + testCases := []struct { + name string + err error + expected bool + }{ + { + name: "context canceled", + err: context.Canceled, + expected: true, + }, + { + name: "eof", + err: io.EOF, + expected: true, + }, + { + name: "grpc canceled", + err: status.Error(codes.Canceled, "stream canceled"), + expected: true, + }, + { + name: "grpc unavailable eof", + err: status.Error(codes.Unavailable, "error reading from server: EOF"), + expected: true, + }, + { + name: "grpc unavailable transport closing", + err: status.Error(codes.Unavailable, "transport is closing"), + expected: true, + }, + { + name: "grpc unavailable client closing", + err: status.Error(codes.Unavailable, "client connection is closing"), + expected: true, + }, + { + name: "non status error", + err: errors.New("stream failed"), + }, + { + name: "grpc unavailable other", + err: status.Error(codes.Unavailable, "connection refused"), + }, + { + name: "grpc internal eof", + err: status.Error(codes.Internal, "EOF"), + }, + } + + for _, tc := range testCases { + t.Run(tc.name, func(t *testing.T) { + require.Equal(t, tc.expected, isExpectedStreamClose(tc.err)) + }) + } +} + +func packPublishEventForClientTest(t *testing.T, event proto.Message) []byte { + t.Helper() + + eventAny, err := codectypes.NewAnyWithValue(event) + require.NoError(t, err) + eventBz, err := proto.Marshal(eventAny) + require.NoError(t, err) + + return eventBz +} + +func newTestPublishEventDecoder(t *testing.T) *publishEventDecoder { + t.Helper() + + decoder, err := newPublishEventDecoder() + require.NoError(t, err) + return decoder +} diff --git a/client/mq/stream/server/config.go b/client/mq/stream/server/config.go new file mode 100644 index 000000000..8b835d719 --- /dev/null +++ b/client/mq/stream/server/config.go @@ -0,0 +1,67 @@ +package main + +import ( + "errors" + "fmt" + "strings" + "time" +) + +const ( + flagMQStreamListenAddress = "listen-address" + flagMQStreamKafkaBrokers = "kafka-brokers" + flagMQStreamEnforceKeepalive = "mq-enforce-keepalive" + flagMQStreamMinClientPingInterval = "mq-min-client-ping-interval" + flagMQStreamMaxConnectionIdle = "mq-max-connection-idle" + flagMQStreamServerPingInterval = "mq-server-ping-interval" + flagMQStreamServerPingResponseTimeout = "mq-server-ping-response-timeout" + + defaultMQStreamMinClientPingInterval = 30 * time.Second + defaultMQStreamMaxConnectionIdle = 180 * time.Second + defaultMQStreamServerPingInterval = 60 * time.Second + defaultMQStreamServerPingResponseTimeout = 40 * time.Second +) + +type mqStreamConfig struct { + ListenAddress string + KafkaBrokers []string + EnforceKeepalive bool + MinClientPingInterval time.Duration + MaxConnectionIdle time.Duration + ServerPingInterval time.Duration + ServerPingResponseTimeout time.Duration +} + +func (cfg mqStreamConfig) Validate() error { + if strings.TrimSpace(cfg.ListenAddress) == "" { + return errors.New("invalid MQ stream config: listen address cannot be empty") + } + + if len(cfg.KafkaBrokers) == 0 { + return errors.New("invalid MQ stream config: no Kafka brokers specified") + } + + for i, broker := range cfg.KafkaBrokers { + if strings.TrimSpace(broker) == "" { + return fmt.Errorf("invalid MQ stream config: Kafka broker #%d is empty", i+1) + } + } + + if cfg.MinClientPingInterval <= 0 { + return errors.New("invalid MQ stream config: min client ping interval must be positive") + } + + if cfg.MaxConnectionIdle <= 0 { + return errors.New("invalid MQ stream config: max connection idle must be positive") + } + + if cfg.ServerPingInterval <= 0 { + return errors.New("invalid MQ stream config: server ping interval must be positive") + } + + if cfg.ServerPingResponseTimeout <= 0 { + return errors.New("invalid MQ stream config: server ping response timeout must be positive") + } + + return nil +} diff --git a/client/mq/stream/server/main.go b/client/mq/stream/server/main.go new file mode 100644 index 000000000..30bcf5e01 --- /dev/null +++ b/client/mq/stream/server/main.go @@ -0,0 +1,193 @@ +package main + +import ( + "context" + "errors" + "flag" + "fmt" + "net" + "os" + "os/signal" + "strings" + "syscall" + + sdklog "cosmossdk.io/log" + "github.com/cosmos/cosmos-sdk/server/grpc/gogoreflection" + "github.com/twmb/franz-go/pkg/kgo" + "google.golang.org/grpc" + "google.golang.org/grpc/codes" + "google.golang.org/grpc/keepalive" + "google.golang.org/grpc/status" + + "github.com/InjectiveLabs/sdk-go/chain/mq/types" +) + +func main() { + var ( + listenAddress = flag.String(flagMQStreamListenAddress, "0.0.0.0:9988", "Listen address") + kafkaBrokers = flag.String(flagMQStreamKafkaBrokers, "", "Comma-separated Kafka broker addresses") //nolint // it's fine + enforceKeepalive = flag.Bool(flagMQStreamEnforceKeepalive, false, "Define if Keepalive configuration params should be applied to MQ event stream gRPC server") //nolint // it's fine + minClientPingInterval = flag.Duration(flagMQStreamMinClientPingInterval, defaultMQStreamMinClientPingInterval, "Duration a client should wait before sending a keepalive ping") //nolint // it's fine + maxConnectionIdle = flag.Duration(flagMQStreamMaxConnectionIdle, defaultMQStreamMaxConnectionIdle, "Duration a connection is allowed to stay idle before forcing the disconnection") //nolint // it's fine + serverPingInterval = flag.Duration(flagMQStreamServerPingInterval, defaultMQStreamServerPingInterval, "Duration after which the server will send a keepalive ping to the client on an idle connection") //nolint // it's fine + serverPingResponseTimeout = flag.Duration(flagMQStreamServerPingResponseTimeout, defaultMQStreamServerPingResponseTimeout, "Duration the server waits for the client to respond to a ping message before forcing a disconnection") //nolint // it's fine + ) + + flag.Parse() + + cfg := mqStreamConfig{ + ListenAddress: *listenAddress, + EnforceKeepalive: *enforceKeepalive, + MinClientPingInterval: *minClientPingInterval, + MaxConnectionIdle: *maxConnectionIdle, + ServerPingInterval: *serverPingInterval, + ServerPingResponseTimeout: *serverPingResponseTimeout, + } + + if *kafkaBrokers != "" { + cfg.KafkaBrokers = strings.Split(*kafkaBrokers, ",") + } + + if err := startMQStream(context.Background(), cfg); err != nil && !errors.Is(err, context.Canceled) { + _, _ = fmt.Fprintln(os.Stderr, err) + os.Exit(1) + } +} + +func startMQStream(ctx context.Context, cfg mqStreamConfig) error { + if err := cfg.Validate(); err != nil { + return err + } + + signalCtx, stopSignals := signal.NotifyContext(ctx, os.Interrupt, syscall.SIGTERM) + defer stopSignals() + + ctx, cancel := context.WithCancelCause(signalCtx) + defer cancel(nil) + + var grpcServerOptions []grpc.ServerOption + if cfg.EnforceKeepalive { + grpcServerOptions = []grpc.ServerOption{ + grpc.KeepaliveEnforcementPolicy(keepalive.EnforcementPolicy{ + MinTime: cfg.MinClientPingInterval, + }), + grpc.KeepaliveParams(keepalive.ServerParameters{ + MaxConnectionIdle: cfg.MaxConnectionIdle, + Time: cfg.ServerPingInterval, + Timeout: cfg.ServerPingResponseTimeout, + }), + } + } + + logger := sdklog.NewLogger(os.Stderr) + srv := NewStreamServer(logger, cfg) + grpcServer := grpc.NewServer(grpcServerOptions...) + types.RegisterEventStreamServer(grpcServer, srv) + gogoreflection.Register(grpcServer) + + listener, err := net.Listen("tcp", strings.TrimPrefix(cfg.ListenAddress, "tcp://")) + if err != nil { + return err + } + + defer func() { + _ = listener.Close() + }() + + serveErrCh := make(chan error, 1) + go func() { + serveErr := grpcServer.Serve(listener) + if errors.Is(serveErr, grpc.ErrServerStopped) { + serveErr = nil + } + + serveErrCh <- serveErr + }() + + logger.Info("event stream server started", "address", cfg.ListenAddress) + + select { + case <-ctx.Done(): + logger.Info("stopping event stream server") + grpcServer.Stop() + if err := context.Cause(ctx); err != nil && !errors.Is(err, context.Canceled) { + return err + } + + return nil + + case err := <-serveErrCh: + cancel(err) + return err + } +} + +type StreamServer struct { + logger sdklog.Logger + kafkaBrokers []string +} + +func NewStreamServer(l sdklog.Logger, cfg mqStreamConfig) *StreamServer { + return &StreamServer{ + logger: l, + kafkaBrokers: cfg.KafkaBrokers, + } +} + +func (srv *StreamServer) EventStream(req *types.EventStreamRequest, server types.EventStream_EventStreamServer) error { + if req == nil { + return status.Error(codes.InvalidArgument, "event stream request cannot be nil") + } + + consumerID := strings.TrimSpace(req.ConsumerId) + if consumerID == "" { + return status.Error(codes.InvalidArgument, "consumer id cannot be empty") + } + + topic := strings.TrimSpace(req.Topic) + if topic == "" { + return status.Error(codes.InvalidArgument, "topic cannot be empty") + } + + opts := []kgo.Opt{ + kgo.SeedBrokers(srv.kafkaBrokers...), + kgo.ConsumerGroup(consumerID), + kgo.ConsumeTopics(topic), + kgo.ConsumeResetOffset(kgo.NewOffset().AtStart()), + } + + client, err := kgo.NewClient(opts...) + if err != nil { + return err + } + + defer client.Close() + + srv.logger.Info("event stream consumer started", "consumer_id", consumerID, "topic", topic) + defer srv.logger.Info("event stream consumer stopped", "consumer_id", consumerID, "topic", topic) + + for { + fetches := client.PollFetches(server.Context()) + if err := fetches.Err(); err != nil { + if errors.Is(err, context.Canceled) || errors.Is(server.Context().Err(), context.Canceled) { + return server.Context().Err() + } + + return fmt.Errorf("poll kafka fetches: %w", err) + } + + iter := fetches.RecordIter() + for !iter.Done() { + record := iter.Next() + var msg types.EventStreamResponse + if err := msg.Unmarshal(record.Value); err != nil { + srv.logger.Error("error decoding event message", "err", err.Error()) + continue + } + + if err := server.Send(&msg); err != nil { + return fmt.Errorf("error sending message to client: %w", err) + } + } + } +} diff --git a/go.mod b/go.mod index eadab7cc6..e18fdfa58 100644 --- a/go.mod +++ b/go.mod @@ -5,6 +5,7 @@ go 1.26.4 require ( cosmossdk.io/api v0.7.6 cosmossdk.io/errors v1.0.1 + cosmossdk.io/log v1.4.1 cosmossdk.io/math v1.4.0 cosmossdk.io/store v1.1.1 cosmossdk.io/x/evidence v0.1.1 @@ -17,6 +18,7 @@ require ( github.com/btcsuite/btcd/btcutil v1.1.6 github.com/cardano-foundation/cardano-ibc-incubator/cosmos/cardano-probabilistic-light-client-v8 v0.1.3 github.com/cometbft/cometbft v1.0.1 + github.com/cometbft/cometbft/api v1.0.0 github.com/cosmos/cosmos-proto v1.0.0-beta.5 github.com/cosmos/cosmos-sdk v0.50.14 github.com/cosmos/go-bip39 v1.0.0 @@ -33,6 +35,7 @@ require ( github.com/shopspring/decimal v1.2.0 github.com/stretchr/testify v1.11.1 github.com/tidwall/btree v1.7.0 + github.com/twmb/franz-go v1.19.4 github.com/tyler-smith/go-bip39 v1.1.0 golang.org/x/crypto v0.41.0 google.golang.org/genproto/googleapis/api v0.0.0-20250825161204-c5933d9347a5 @@ -47,7 +50,6 @@ require ( cosmossdk.io/collections v0.4.0 // indirect cosmossdk.io/core v0.12.1-0.20231114100755-569e3ff6a0d7 // indirect cosmossdk.io/depinject v1.1.0 // indirect - cosmossdk.io/log v1.4.1 // indirect filippo.io/edwards25519 v1.1.0 // indirect github.com/99designs/go-keychain v0.0.0-20191008050251-8e49817e8af4 // indirect github.com/99designs/keyring v1.2.2 // indirect @@ -81,7 +83,6 @@ require ( github.com/cockroachdb/tokenbucket v0.0.0-20230807174530-cc333fc44b06 // indirect github.com/coder/websocket v1.8.13 // indirect github.com/cometbft/cometbft-db v1.0.4 // indirect - github.com/cometbft/cometbft/api v1.0.0 // indirect github.com/consensys/gnark-crypto v0.18.0 // indirect github.com/cosmos/btcutil v1.0.5 // indirect github.com/cosmos/cosmos-db v1.1.1 // indirect @@ -171,6 +172,7 @@ require ( github.com/opencontainers/go-digest v1.0.0 // indirect github.com/pelletier/go-toml/v2 v2.2.3 // indirect github.com/petermattis/goid v0.0.0-20240813172612-4fcff4a6cae7 // indirect + github.com/pierrec/lz4/v4 v4.1.22 // indirect github.com/pmezard/go-difflib v1.0.1-0.20181226105442-5d4384ee4fb2 // indirect github.com/prometheus/client_golang v1.21.1 // indirect github.com/prometheus/client_model v0.6.1 // indirect @@ -199,6 +201,7 @@ require ( github.com/tendermint/go-amino v0.16.0 // indirect github.com/tklauser/go-sysconf v0.3.15 // indirect github.com/tklauser/numcpus v0.10.0 // indirect + github.com/twmb/franz-go/pkg/kmsg v1.11.2 // indirect github.com/utxorpc/go-codegen v0.5.1 // indirect github.com/x448/float16 v0.8.4 // indirect github.com/yusufpapurcu/wmi v1.2.4 // indirect diff --git a/go.sum b/go.sum index c41a679b2..a9c5e4eef 100644 --- a/go.sum +++ b/go.sum @@ -1979,6 +1979,8 @@ github.com/phpdave11/gofpdf v1.4.2/go.mod h1:zpO6xFn9yxo3YLyMvW8HcKWVdbNqgIfOOp2 github.com/phpdave11/gofpdi v1.0.12/go.mod h1:vBmVV0Do6hSBHC8uKUQ71JGW+ZGQq74llk/7bXwjDoI= github.com/phpdave11/gofpdi v1.0.13/go.mod h1:vBmVV0Do6hSBHC8uKUQ71JGW+ZGQq74llk/7bXwjDoI= github.com/pierrec/lz4/v4 v4.1.15/go.mod h1:gZWDp/Ze/IJXGXf23ltt2EXimqmTUXEy0GFuRQyBid4= +github.com/pierrec/lz4/v4 v4.1.22 h1:cKFw6uJDK+/gfw5BcDL0JL5aBsAFdsIT18eRtLj7VIU= +github.com/pierrec/lz4/v4 v4.1.22/go.mod h1:gZWDp/Ze/IJXGXf23ltt2EXimqmTUXEy0GFuRQyBid4= github.com/pingcap/errors v0.11.4 h1:lFuQV/oaUMGcD2tqt+01ROSmJs75VG1ToEOkZIZ4nE4= github.com/pingcap/errors v0.11.4/go.mod h1:Oi8TUi2kEtXXLMJk9l1cGmz20kV3TaQ0usTwv5KuLY8= github.com/pion/dtls/v2 v2.2.7 h1:cSUBsETxepsCSFSxC3mc/aDo14qQLMSL+O6IjG28yV8= @@ -2122,6 +2124,10 @@ github.com/tklauser/go-sysconf v0.3.15/go.mod h1:Dmjwr6tYFIseJw7a3dRLJfsHAMXZ3nE github.com/tklauser/numcpus v0.10.0 h1:18njr6LDBk1zuna922MgdjQuJFjrdppsZG60sHGfjso= github.com/tklauser/numcpus v0.10.0/go.mod h1:BiTKazU708GQTYF4mB+cmlpT2Is1gLk7XVuEeem8LsQ= github.com/tv42/httpunix v0.0.0-20150427012821-b75d8614f926/go.mod h1:9ESjWnEqriFuLhtthL60Sar/7RFoluCcXsuvEwTV5KM= +github.com/twmb/franz-go v1.19.4 h1:0ktflzm5YU7+YYdie8RQWFcU9uDJ03xLefplO1iMwO4= +github.com/twmb/franz-go v1.19.4/go.mod h1:4kFJ5tmbbl7asgwAGVuyG1ZMx0NNpYk7EqflvWfPCpM= +github.com/twmb/franz-go/pkg/kmsg v1.11.2 h1:hIw75FpwcAjgeyfIGFqivAvwC5uNIOWRGvQgZhH4mhg= +github.com/twmb/franz-go/pkg/kmsg v1.11.2/go.mod h1:CFfkkLysDNmukPYhGzuUcDtf46gQSqCZHMW1T4Z+wDE= github.com/ulikunitz/xz v0.5.11 h1:kpFauv27b6ynzBNT/Xy+1k+fK4WswhN/6PN5WhFAGw8= github.com/ulikunitz/xz v0.5.11/go.mod h1:nbz6k7qbPmH4IRqmfOplQw/tblSgqTqBwxkY0oWt/14= github.com/urfave/cli/v2 v2.27.5 h1:WoHEJLdsXr6dDWoJgMq/CboDmyY/8HMMH1fTECbih+w= diff --git a/proto/buf.lock b/proto/buf.lock index ce82fa77b..9647b6490 100644 --- a/proto/buf.lock +++ b/proto/buf.lock @@ -1,6 +1,11 @@ # Generated by buf. DO NOT EDIT. version: v1 deps: + - remote: buf.build + owner: cometbft + repository: cometbft + commit: 68e42a58b3014033aee53aa3202bf37c + digest: shake256:51f2fc29511edaadb51b5935a979d5c8e699deed4fe9509180ad3e5fc46f390200740d0cc4781300e927caf24e0079804218c21e6426b4373adfa10a2f59edae - remote: buf.build owner: cosmos repository: cosmos-proto @@ -19,15 +24,15 @@ deps: - remote: buf.build owner: cosmwasm repository: wasmd - commit: 3c30b822226c4bea96d5fad293a6e010 - digest: shake256:6c4ea55d553dbad8f7ac9b5d91f2a784742a1d75783e690b69edd5a72bf31abbcbb7556fa4da6f38de22d58f3211e738f9b37bc39fc502d8d8a7b290e3235bb7 + commit: 65537c618a924482b0d5ed51230d92dd + digest: shake256:88e9c6f2d91919064b55bcfd7af69086cfa687065666526dba91c7fe0428135c48b009c9768f3ff5fa07eefaccedbab0e3268faaff6619d4464038004218234e - remote: buf.build owner: googleapis repository: googleapis - commit: 61b203b9a9164be9a834f58c37be6f62 - digest: shake256:e619113001d6e284ee8a92b1561e5d4ea89a47b28bf0410815cb2fa23914df8be9f1a6a98dcf069f5bc2d829a2cfb1ac614863be45cd4f8a5ad8606c5f200224 + commit: c17df5b2beca46928cc87d5656bd5343 + digest: shake256:c62ecead9b13485a02893cd678a6c81e40879bf00ea509bbc6fd8f1b2cc33eccf6a85c259b08d1e0f052f693cbfc7dfda236e9665b1d6869b8e1132a794a61e2 - remote: buf.build owner: protocolbuffers repository: wellknowntypes - commit: 7727a3b7399540398e5736b9916f0faa - digest: shake256:74c046f009811965d227acd11bb2c6259bf387759d4230b95405e402a2ed4212d6f4babab690407d07bc81095c3917c4ae9144195ddfb081525c9aec58e51173 + commit: 122a3d2fdc814e0bb6fcf821bcce7957 + digest: shake256:e7579d76656bf352b2369d1c855f3a3ce25aeb3fd4a0505ad065e468fab2df935d6f2d14daa7d5dd98b4cd8102c6a509ede5969783be083b0e35e279b5de8be6 diff --git a/proto/buf.yaml b/proto/buf.yaml index 65795cec3..5769d7e20 100644 --- a/proto/buf.yaml +++ b/proto/buf.yaml @@ -7,6 +7,7 @@ deps: - buf.build/cosmos/gogo-proto - buf.build/cosmwasm/wasmd:main - buf.build/protocolbuffers/wellknowntypes + - buf.build/cometbft/cometbft breaking: use: - FILE diff --git a/proto/injective/mq/v1beta1/query.proto b/proto/injective/mq/v1beta1/query.proto new file mode 100644 index 000000000..7f8c7d942 --- /dev/null +++ b/proto/injective/mq/v1beta1/query.proto @@ -0,0 +1,52 @@ +syntax = "proto3"; +package injective.mq.v1beta1; + +import "gogoproto/gogo.proto"; +import "cometbft/abci/v1/types.proto"; + +option go_package = "github.com/InjectiveLabs/injective-core/injective-chain/mq/types"; + +service EventStream { + rpc EventStream(EventStreamRequest) returns (stream EventStreamResponse); +} + +message EventStreamRequest { + // a unique Kafka partition for this particular client. Reuse the value to + // resume where you left off with receiving data + string consumer_id = 1; + + // topic of interest (raw, latest, guaranteed) + string topic = 2; +} + +message EventStreamResponse { + int64 block_height = 1; + bytes app_hash = 2; + bytes last_app_hash = 3; + EventSet block_events = 4 [ (gogoproto.nullable) = false ]; + repeated EventSet tx_events = 5 [ (gogoproto.nullable) = false ]; + + repeated AdditionalDataEntry additional_data_entries = 6; +} + +// todo: still needed? +enum AdditionalDataTypes { + EVM = 0; +} + +message AdditionalDataEntry { + AdditionalDataTypes type = 1; + bytes data = 2; +} + +enum EventType { + ABCI = 0; + PUBLISH = 1; +} + +message EventSet { + repeated bytes published_events = 1; + repeated EventType true_orders = 2; + repeated cometbft.abci.v1.Event abci_events = 3 + [ (gogoproto.nullable) = false ]; +}