committer redo #287
There are no files selected for viewing
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,29 @@ | ||
| package cmd | ||
|
|
||
| import ( | ||
| "fmt" | ||
|
|
||
| "github.com/rs/zerolog/log" | ||
| "github.com/spf13/cobra" | ||
| "github.com/thirdweb-dev/indexer/internal/committer" | ||
| "github.com/thirdweb-dev/indexer/internal/rpc" | ||
| ) | ||
|
|
||
| var committerCmd = &cobra.Command{ | ||
| Use: "committer", | ||
| Short: "run committer", | ||
| Long: "published data from s3 to kafka. if block is not found in s3, it will panic", | ||
| Run: RunCommitter, | ||
| } | ||
|
|
||
| func RunCommitter(cmd *cobra.Command, args []string) { | ||
| fmt.Println("running committer") | ||
| rpc, err := rpc.Initialize() | ||
| if err != nil { | ||
| log.Fatal().Err(err).Msg("Failed to initialize RPC") | ||
| } | ||
| chainId := rpc.GetChainID() | ||
|
|
||
| committer.Init(chainId, rpc) | ||
| committer.Commit(chainId) | ||
| } | ||
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,180 @@ | ||
| # Committer Package | ||
|
|
||
| This package implements a committer that processes block data from S3 parquet files and publishes them to Kafka. It follows the requirements specified in the original comments. | ||
|
|
||
| ## Features | ||
|
|
||
| - **ClickHouse Integration**: Gets the maximum block number from ClickHouse for the chain | ||
| - **S3 File Discovery**: Lists parquet files from S3 with chain-specific prefixes | ||
| - **Block Range Parsing**: Extracts start and end block numbers from S3 filenames | ||
| - **File Filtering**: Skips files where end block is less than max block number from ClickHouse | ||
| - **Sequential Processing**: Processes files in ascending order by start block number | ||
| - **Memory-Efficient Streaming**: Streams parquet files row-by-row to minimize memory usage | ||
| - **Kafka Publishing**: Publishes processed block data to Kafka | ||
| - **Error Handling**: Comprehensive error handling with detailed logging | ||
|
|
||
| ## Usage | ||
|
|
||
| ### Basic Usage | ||
|
|
||
| ```go | ||
| package main | ||
|
|
||
| import ( | ||
| "context" | ||
| "math/big" | ||
| "log" | ||
|
|
||
| "github.com/thirdweb-dev/indexer/internal/committer" | ||
| "github.com/thirdweb-dev/indexer/configs" | ||
| ) | ||
|
|
||
| func main() { | ||
| // Load configuration | ||
| if err := configs.LoadConfig("config.yml"); err != nil { | ||
| log.Fatal("Failed to load config:", err) | ||
| } | ||
|
|
||
| // Create committer for chain ID 1 (Ethereum mainnet) | ||
| chainId := big.NewInt(1) | ||
| committer, err := committer.NewCommitterFromConfig(chainId) | ||
| if err != nil { | ||
| log.Fatal("Failed to create committer:", err) | ||
| } | ||
| defer committer.Close() | ||
|
|
||
| // Process blocks | ||
| ctx := context.Background() | ||
| if err := committer.ProcessBlocks(ctx); err != nil { | ||
| log.Fatal("Failed to process blocks:", err) | ||
| } | ||
| } | ||
| ``` | ||
|
Comment on lines
+20
to
+52
There was a problem hiding this comment. Examples reference non-existent APIs; align with actual public surface. README shows NewCommitterFromConfig/NewCommitter/ProcessBlocks, but the code exposes Init/Commit (and a CLI subcommand). Please update the examples to either:
I can draft a corrected snippet once you confirm the intended public API. Also applies to: 56-104 |
||
|
|
||
| ### Advanced Usage with Custom Configuration | ||
|
|
||
| ```go | ||
| package main | ||
|
|
||
| import ( | ||
| "context" | ||
| "math/big" | ||
| "log" | ||
|
|
||
| "github.com/thirdweb-dev/indexer/internal/committer" | ||
| "github.com/thirdweb-dev/indexer/configs" | ||
| ) | ||
|
|
||
| func main() { | ||
| // Custom configuration | ||
| chainId := big.NewInt(137) // Polygon | ||
|
|
||
| clickhouseConfig := &configs.ClickhouseConfig{ | ||
| Host: "localhost", | ||
| Port: 9000, | ||
| Username: "default", | ||
| Password: "", | ||
| Database: "insight", | ||
| } | ||
|
|
||
| s3Config := &configs.S3Config{ | ||
| Bucket: "thirdweb-insight-production", | ||
| Region: "us-east-1", | ||
| AccessKeyID: "your-access-key", | ||
| SecretAccessKey: "your-secret-key", | ||
| } | ||
|
|
||
| kafkaConfig := &configs.KafkaConfig{ | ||
| Brokers: "localhost:9092", | ||
| } | ||
|
|
||
| // Create committer | ||
| committer, err := committer.NewCommitter(chainId, clickhouseConfig, s3Config, kafkaConfig) | ||
| if err != nil { | ||
| log.Fatal("Failed to create committer:", err) | ||
| } | ||
| defer committer.Close() | ||
|
|
||
| // Process blocks | ||
| ctx := context.Background() | ||
| if err := committer.ProcessBlocks(ctx); err != nil { | ||
| log.Fatal("Failed to process blocks:", err) | ||
| } | ||
| } | ||
| ``` | ||
|
|
||
| ## Configuration Requirements | ||
|
|
||
| The committer requires the following configuration: | ||
|
|
||
| ### ClickHouse Configuration | ||
| - Host, Port, Username, Password, Database | ||
| - Used to query the maximum block number for the chain | ||
|
|
||
| ### S3 Configuration | ||
| - Bucket name (e.g., "thirdweb-insight-production") | ||
| - Region, Access Key ID, Secret Access Key | ||
| - Used to list and download parquet files | ||
|
|
||
| ### Kafka Configuration | ||
| - Brokers list | ||
| - Used to publish processed block data | ||
|
|
||
| ## S3 File Structure | ||
|
|
||
| The committer expects S3 files to follow this naming pattern: | ||
| ``` | ||
| chain_${chainId}/year=2024/blocks_1000_2000.parquet | ||
| ``` | ||
|
|
||
| Where: | ||
| - `chain_${chainId}` is the prefix for the chain | ||
| - `year=2024` is the partitioning by year | ||
| - `blocks_1000_2000.parquet` contains blocks from 1000 to 2000 | ||
|
|
||
| ## Parquet File Structure | ||
|
|
||
| The parquet files should contain the following columns: | ||
| - `chain_id` (uint64): Chain identifier | ||
| - `block_number` (uint64): Block number | ||
| - `block_hash` (string): Block hash | ||
| - `block_timestamp` (int64): Block timestamp | ||
| - `block_json` (bytes): Serialized block data | ||
| - `transactions_json` (bytes): Serialized transactions data | ||
| - `logs_json` (bytes): Serialized logs data | ||
| - `traces_json` (bytes): Serialized traces data | ||
|
|
||
| ## Processing Flow | ||
|
|
||
| 1. **Query ClickHouse**: Get the maximum block number for the chain | ||
| 2. **List S3 Files**: Find all parquet files with the chain prefix | ||
| 3. **Filter Files**: Skip files where end block ≤ max block number | ||
| 4. **Sort Files**: Order by start block number (ascending) | ||
| 5. **Process Sequentially**: For each file: | ||
| - Download from S3 to local storage | ||
| - Stream parquet file row-by-row | ||
| - Skip blocks < next commit block number | ||
| - Error if block > next commit block number (missing data) | ||
| - Publish found blocks to Kafka | ||
| - Increment commit block number | ||
| - Clean up local file | ||
|
|
||
| ## Error Handling | ||
|
|
||
| The committer includes comprehensive error handling: | ||
| - Missing configuration validation | ||
| - S3 connection and download errors | ||
| - Parquet file parsing errors | ||
| - Kafka publishing errors | ||
| - Block sequence validation errors | ||
|
|
||
| All errors are logged with detailed context for debugging. | ||
|
|
||
| ## Memory Management | ||
|
|
||
| The committer is designed to be memory-efficient: | ||
| - Downloads files directly to disk (no in-memory buffering) | ||
| - Streams parquet files row-by-row | ||
| - Processes one file at a time | ||
| - Cleans up local files after processing | ||
| - Uses semaphores to limit concurrent operations | ||
Oops, something went wrong.
Oops, something went wrong.
Add this suggestion to a batch that can be applied as a single commit.
This suggestion is invalid because no changes were made to the code.
Suggestions cannot be applied while the pull request is closed.
Suggestions cannot be applied while viewing a subset of changes.
Only one suggestion per line can be applied in a batch.
Add this suggestion to a batch that can be applied as a single commit.
Applying suggestions on deleted lines is not supported.
You must change the existing code in this line in order to create a valid suggestion.
Outdated suggestions cannot be applied.
This suggestion has been applied or marked resolved.
Suggestions cannot be applied from pending reviews.
Suggestions cannot be applied on multi-line comments.
Suggestions cannot be applied while the pull request is queued to merge.
Suggestion cannot be applied right now. Please check back later.
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Add proper error handling and resource cleanup.
The function should properly handle errors and clean up resources when the committer finishes or encounters an error.
func RunCommitter(cmd *cobra.Command, args []string) { fmt.Println("running committer") rpc, err := rpc.Initialize() if err != nil { log.Fatal().Err(err).Msg("Failed to initialize RPC") } + defer rpc.Close() + chainId := rpc.GetChainID() committer.Init(chainId, rpc) - committer.Commit(chainId) + if err := committer.Commit(chainId); err != nil { + committer.Close() + log.Fatal().Err(err).Msg("Committer failed") + } + if err := committer.Close(); err != nil { + log.Error().Err(err).Msg("Failed to close committer cleanly") + } }🤖 Prompt for AI Agents