Skip to content
New issue

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

Sign up for GitHub

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

Already on GitHub? Sign in to your account

feat: filterable event streaming #212

Draft
wants to merge 4 commits into
base: master
Choose a base branch
from
Draft

feat: filterable event streaming #212

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
4 commits
Select commit Hold shift + click to select a range
dd2a1c0
feat: filterable event streaming
seanmcgary Jan 28, 2025
7377a21
wip
seanmcgary Feb 4, 2025
463af83
parse filters and filter results
seanmcgary Feb 13, 2025
806ce1c
set EigenState struct fields as filterable
seanmcgary Feb 13, 2025
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
5 changes: 4 additions & 1 deletion cmd/debugger/main.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -120,10 +120,13 @@ func main() {
GenesisBlockNumber: cfg.GetGenesisBlockNumber(),
}, cfg, mds, p, sm, msm, rc, rcq, rps, l, client)

rpc := rpcServer.NewRpcServer(&rpcServer.RpcServerConfig{
rpc, err := rpcServer.NewRpcServer(&rpcServer.RpcServerConfig{
GrpcPort: cfg.RpcConfig.GrpcPort,
HttpPort: cfg.RpcConfig.HttpPort,
}, mds, rc, rcq, eb, rps, pds, rds, scc, l, cfg)
if err != nil {
l.Sugar().Fatalw("Failed to create rpc server", zap.Error(err))
}

// RPC channel to notify the RPC server to shutdown gracefully
rpcChannel := make(chan bool)
Expand Down
6 changes: 5 additions & 1 deletion cmd/rpc.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -109,11 +109,15 @@ var rpcCmd = &cobra.Command{
l.Sugar().Fatalw("Failed to create sidecar client", zap.Error(err))
}

rpc := rpcServer.NewRpcServer(&rpcServer.RpcServerConfig{
rpc, err := rpcServer.NewRpcServer(&rpcServer.RpcServerConfig{
GrpcPort: cfg.RpcConfig.GrpcPort,
HttpPort: cfg.RpcConfig.HttpPort,
}, mds, rc, rcq, eb, rps, pds, rds, scc, l, cfg)

if err != nil {
l.Sugar().Fatalw("Failed to create rpc server", zap.Error(err))
}

// RPC channel to notify the RPC server to shutdown gracefully
rpcChannel := make(chan bool)
if err := rpc.Start(ctx, rpcChannel); err != nil {
Expand Down
5 changes: 4 additions & 1 deletion cmd/run.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -143,10 +143,13 @@ var runCmd = &cobra.Command{
GenesisBlockNumber: cfg.GetGenesisBlockNumber(),
}, cfg, mds, p, sm, msm, rc, rcq, rps, l, client)

rpc := rpcServer.NewRpcServer(&rpcServer.RpcServerConfig{
rpc, err := rpcServer.NewRpcServer(&rpcServer.RpcServerConfig{
GrpcPort: cfg.RpcConfig.GrpcPort,
HttpPort: cfg.RpcConfig.HttpPort,
}, mds, rc, rcq, eb, rps, pds, rds, scc, l, cfg)
if err != nil {
l.Sugar().Fatalw("Failed to create rpc server", zap.Error(err))
}

// RPC channel to notify the RPC server to shutdown gracefully
rpcChannel := make(chan bool)
Expand Down
3 changes: 2 additions & 1 deletion examples/eventSubscriber/main.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -35,6 +35,7 @@ func streamIndexedBlocks(client v1.EventsClient) {
panic(err)
}

fmt.Printf("Client has connected to server\n")
for {
resp := &v1.StreamIndexedBlocksResponse{}
err := stream.RecvMsg(resp)
Expand Down Expand Up @@ -77,5 +78,5 @@ func main() {
panic(err)
}

streamStateChanges(client)
streamIndexedBlocks(client)
}
2 changes: 1 addition & 1 deletion go.mod
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -6,7 +6,7 @@ require (
github.com/DataDog/datadog-go/v5 v5.5.0
github.com/Layr-Labs/eigenlayer-contracts v0.4.1-holesky-pepe.0.20240813143901-00fc4b95e9c1
github.com/Layr-Labs/eigenlayer-rewards-proofs v0.2.13
github.com/Layr-Labs/protocol-apis v1.7.0
github.com/Layr-Labs/protocol-apis v1.7.1-0.20250213193904-02e6c9a33dbf
github.com/ethereum/go-ethereum v1.14.13
github.com/gocarina/gocsv v0.0.0-20240520201108-78e41c74b4b1
github.com/google/uuid v1.6.0
Expand Down
8 changes: 4 additions & 4 deletions go.sum
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -43,12 +43,12 @@ github.com/Layr-Labs/eigenlayer-contracts v0.4.1-holesky-pepe.0.20240813143901-0
github.com/Layr-Labs/eigenlayer-contracts v0.4.1-holesky-pepe.0.20240813143901-00fc4b95e9c1/go.mod h1:Ie8YE3EQkTHqG6/tnUS0He7/UPMkXPo/3OFXwSy0iRo=
github.com/Layr-Labs/eigenlayer-rewards-proofs v0.2.13 h1:Blb4AE+jC/vddV71w4/MQAPooM+8EVqv9w2bL4OytgY=
github.com/Layr-Labs/eigenlayer-rewards-proofs v0.2.13/go.mod h1:PD/HoyzZjxDw1tAcZw3yD0yGddo+yhmwQAi+lk298r4=
github.com/Layr-Labs/protocol-apis v1.6.0 h1:1TTi4+t8Kq4YKsji7xa3CPGYf4fHXZZbbpt3GYQWZ78=
github.com/Layr-Labs/protocol-apis v1.6.0/go.mod h1:zCirDItAbrnEv1kV1RTccY7eVSg0+da4/dFCXHyLNZQ=
github.com/Layr-Labs/protocol-apis v1.6.1-0.20250212194100-39ec43be3227 h1:erE0SEuSNrYMf5Hq7cgvDdJTunN3KnkDJgwPngQpWbc=
github.com/Layr-Labs/protocol-apis v1.6.1-0.20250212194100-39ec43be3227/go.mod h1:zCirDItAbrnEv1kV1RTccY7eVSg0+da4/dFCXHyLNZQ=
github.com/Layr-Labs/protocol-apis v1.7.0 h1:BYzfF0mLjil3VAq8EEnswJbOL9v0OqfL32TipUryZ2s=
github.com/Layr-Labs/protocol-apis v1.7.0/go.mod h1:zCirDItAbrnEv1kV1RTccY7eVSg0+da4/dFCXHyLNZQ=
github.com/Layr-Labs/protocol-apis v1.7.1-0.20250213185609-9608503ad4ed h1:dWv+54pLMoDWTPiVWatfFvyv5tLlkyuK01KBAvW4ieY=
github.com/Layr-Labs/protocol-apis v1.7.1-0.20250213185609-9608503ad4ed/go.mod h1:zCirDItAbrnEv1kV1RTccY7eVSg0+da4/dFCXHyLNZQ=
github.com/Layr-Labs/protocol-apis v1.7.1-0.20250213193904-02e6c9a33dbf h1:0ajrteTXg+E6jrkd9bPJkr8LvUTCpzqXUq7fFVPMGL4=
github.com/Layr-Labs/protocol-apis v1.7.1-0.20250213193904-02e6c9a33dbf/go.mod h1:zCirDItAbrnEv1kV1RTccY7eVSg0+da4/dFCXHyLNZQ=
github.com/Microsoft/go-winio v0.5.0/go.mod h1:JPGBdM1cNvN/6ISo+n8V5iA4v8pBzdOpzfwIujj1a84=
github.com/Microsoft/go-winio v0.6.2 h1:F2VQgta7ecxGYO8k3ZZz3RS8fVIXVxONVUPlNERoyfY=
github.com/Microsoft/go-winio v0.6.2/go.mod h1:yd8OoFMLzJbo9gZq8j5qaps8bJ9aShtEA8Ipt1oGCvU=
Expand Down
12 changes: 6 additions & 6 deletions pkg/eigenState/avsOperators/avsOperators.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -17,12 +17,12 @@ import (
)

type AvsOperatorStateChange struct {
Avs string
Operator string
Registered bool
LogIndex uint64
TransactionHash string
BlockNumber uint64
Avs string `filter:"true"`
Operator string `filter:"true"`
Registered bool `filter:"true"`
LogIndex uint64 `filter:"true"`
TransactionHash string `filter:"true"`
BlockNumber uint64 `filter:"true"`
}

// EigenState model for AVS operators that implements IEigenStateModel.
Expand Down
18 changes: 9 additions & 9 deletions pkg/eigenState/operatorShares/operatorShares.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -21,15 +21,15 @@ import (
)

type OperatorShareDeltas struct {
Operator string
Staker string
Strategy string
Shares string
TransactionHash string
LogIndex uint64
BlockNumber uint64
BlockTime time.Time
BlockDate string
Operator string `filter:"true"`
Staker string `filter:"true"`
Strategy string `filter:"true"`
Shares string `filter:"true"`
TransactionHash string `filter:"true"`
LogIndex uint64 `filter:"true"`
BlockNumber uint64 `filter:"true"`
BlockTime time.Time `filter:"true"`
BlockDate string `filter:"true"`
}

func NewSlotID(operator string, strategy string, staker string, transactionHash string, logIndex uint64) types.SlotID {
Expand Down
28 changes: 14 additions & 14 deletions pkg/eigenState/rewardSubmissions/rewardSubmissions.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -20,20 +20,20 @@ import (
)

type RewardSubmission struct {
Avs string
RewardHash string
Token string
Amount string
Strategy string
StrategyIndex uint64
Multiplier string
StartTimestamp *time.Time
EndTimestamp *time.Time
Duration uint64
BlockNumber uint64
RewardType string // avs, all_stakers, all_earners
TransactionHash string
LogIndex uint64
Avs string `filter:"true"`
RewardHash string `filter:"true"`
Token string `filter:"true"`
Amount string `filter:"true"`
Strategy string `filter:"true"`
StrategyIndex uint64 `filter:"true"`
Multiplier string `filter:"true"`
StartTimestamp *time.Time `filter:"true"`
EndTimestamp *time.Time `filter:"true"`
Duration uint64 `filter:"true"`
BlockNumber uint64 `filter:"true"`
RewardType string `filter:"true"` // avs, all_stakers, all_earners
TransactionHash string `filter:"true"`
LogIndex uint64 `filter:"true"`
}

func NewSlotID(transactionHash string, logIndex uint64, rewardHash string, strategyIndex uint64) types.SlotID {
Expand Down
12 changes: 6 additions & 6 deletions pkg/eigenState/stakerDelegations/stakerDelegations.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -17,12 +17,12 @@ import (
)

type StakerDelegationChange struct {
Staker string
Operator string
BlockNumber uint64
Delegated bool
LogIndex uint64
TransactionHash string
Staker string `filter:"true"`
Operator string `filter:"true"`
BlockNumber uint64 `filter:"true"`
Delegated bool `filter:"true"`
LogIndex uint64 `filter:"true"`
TransactionHash string `filter:"true"`
}

type StakerDelegationsModel struct {
Expand Down
20 changes: 10 additions & 10 deletions pkg/eigenState/stakerShares/stakerShares.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -25,16 +25,16 @@ import (

// Table staker_share_deltas
type StakerShareDeltas struct {
Staker string
Strategy string
Shares string
StrategyIndex uint64
TransactionHash string
LogIndex uint64
BlockTime time.Time
BlockDate string
BlockNumber uint64
WithdrawalRootString string `gorm:"-"`
Staker string `filter:"true"`
Strategy string `filter:"true"`
Shares string `filter:"true"`
StrategyIndex uint64 `filter:"true"`
TransactionHash string `filter:"true"`
LogIndex uint64 `filter:"true"`
BlockTime time.Time `filter:"true"`
BlockDate string `filter:"true"`
BlockNumber uint64 `filter:"true"`
WithdrawalRootString string `gorm:"-" filter:"true"`
}

func NewSlotID(transactionHash string, logIndex uint64, staker string, strategy string, strategyIndex uint64) types.SlotID {
Expand Down
28 changes: 14 additions & 14 deletions pkg/eigenState/types/tables.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -5,25 +5,25 @@ import (
)

type SubmittedDistributionRoot struct {
Root string
BlockNumber uint64
RootIndex uint64
RewardsCalculationEnd time.Time
RewardsCalculationEndUnit string
ActivatedAt time.Time
ActivatedAtUnit string
CreatedAtBlockNumber uint64
LogIndex uint64
TransactionHash string
Root string `filter:"true"`
BlockNumber uint64 `filter:"true"`
RootIndex uint64 `filter:"true"`
RewardsCalculationEnd time.Time `filter:"true"`
RewardsCalculationEndUnit string `filter:"true"`
ActivatedAt time.Time `filter:"true"`
ActivatedAtUnit string `filter:"true"`
CreatedAtBlockNumber uint64 `filter:"true"`
LogIndex uint64 `filter:"true"`
TransactionHash string `filter:"true"`
}

func (sdr *SubmittedDistributionRoot) GetSnapshotDate() string {
return sdr.RewardsCalculationEnd.UTC().Format(time.DateOnly)
}

type DisabledDistributionRoot struct {
RootIndex uint64
BlockNumber uint64
LogIndex uint64
TransactionHash string
RootIndex uint64 `filter:"true"`
BlockNumber uint64 `filter:"true"`
LogIndex uint64 `filter:"true"`
TransactionHash string `filter:"true"`
}
Loading
Loading