feat: add publisher mode option #261
Conversation
|
Caution Review failedThe pull request is closed. WalkthroughA new "publisher mode" configuration option was introduced to select between "default" (publish after commit) and "parallel" (publish concurrently with commit) modes. This change adds CLI flags, config fields, and modifies committer logic to run commit and publish loops either sequentially or in parallel. Storage connectors and interfaces were extended to track the last published block number, with corresponding tests and mocks updated. Changes
Sequence Diagram(s)sequenceDiagram
participant User
participant CLI/Config
participant Committer
participant StagingStorage
participant MainStorage
participant Publisher
User->>CLI/Config: Set publisher.mode ("default" or "parallel")
User->>Committer: Start()
alt publisher.mode == "parallel"
Committer->>Committer: runCommitLoop()
Committer->>Committer: runPublishLoop()
Committer->>MainStorage: Commit blockData (in commit loop)
Committer->>StagingStorage: GetLastPublishedBlockNumber(chainId) (in publish loop)
Committer->>Publisher: Publish sequential blocks (in publish loop)
Committer->>StagingStorage: SetLastPublishedBlockNumber(chainId, blockNumber)
else publisher.mode == "default"
User->>Committer: commit(blockData)
Committer->>MainStorage: Commit blockData
Committer->>Publisher: Publish blockData asynchronously
end
Estimated code review effort🎯 3 (Moderate) | ⏱️ ~20 minutes Note 🔌 MCP (Model Context Protocol) integration is now available in Early Access!Pro users can now connect to remote MCP servers under the Integrations page to get reviews and chat conversations that understand additional development context. 📜 Recent review detailsConfiguration used: CodeRabbit UI 📒 Files selected for processing (6)
✨ Finishing Touches
🧪 Generate unit tests
🪧 TipsChatThere are 3 ways to chat with CodeRabbit:
SupportNeed help? Create a ticket on our support page for assistance with any issues or questions. CodeRabbit Commands (Invoked using PR comments)
Other keywords and placeholders
CodeRabbit Configuration File (
|
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
There was a problem hiding this comment.
Actionable comments posted: 5
🔭 Outside diff range comments (1)
configs/config.go (1)
281-283: Bug: Orchestrator chainBasedConfig env override writes into Main storage
STORAGE_ORCHESTRATOR_CLICKHOUSE_CHAINBASEDCONFIGis parsed, but assigned toCfg.Storage.Main.Clickhouseinstead ofCfg.Storage.Orchestrator.Clickhouse.Apply:
- if Cfg.Storage.Main.Clickhouse != nil { - Cfg.Storage.Main.Clickhouse.ChainBasedConfig = orchestratorChainConfig - } + if Cfg.Storage.Orchestrator.Clickhouse != nil { + Cfg.Storage.Orchestrator.Clickhouse.ChainBasedConfig = orchestratorChainConfig + }
🧹 Nitpick comments (6)
configs/config.example.yml (1)
193-197: Nit: clarify fallback behavior in example commentsGiven the implementation falls back to post-commit if the pre-commit cursor lookup fails, consider noting that here to set operator expectations.
For example: “If pre-commit cursor lookup fails, the system automatically falls back to post-commit publishing.”
internal/storage/postgres.go (1)
347-364: Guard against nilchainIdfor robustnessIf
chainIdever propagates as nil,chainId.String()will panic. Add a fast-fail to ease debugging.func (p *PostgresConnector) GetLastPublishedBlockNumber(chainId *big.Int) (*big.Int, error) { + if chainId == nil { + return nil, fmt.Errorf("chainId cannot be nil") + } query := `SELECT cursor_value FROM cursors WHERE cursor_type = 'publish' AND chain_id = $1` ...internal/orchestrator/committer_test.go (2)
355-396: Pre-commit mode test: good async verification; minor improvement possibleUsing a channel signaled by
SetLastPublishedBlockNumberis a solid way to await the async path. Consider also asserting that post-commit publish path is skipped (e.g., by verifying no second cursor update if you later add it there).
397-429: Avoid fixedtime.Sleepto reduce flakinessReplace the fixed
Sleep(100ms)withassert.Eventually(orrequire.Eventually) to robustly wait for "not called" condition without relying on arbitrary timing.- time.Sleep(100 * time.Millisecond) - mockStagingStorage.AssertNotCalled(t, "SetLastPublishedBlockNumber", mock.Anything, mock.Anything) + assert.Eventually(t, func() bool { + // Always true unless the method was called, in which case AssertNotCalled will fail below + return true + }, 2*time.Second, 25*time.Millisecond) + mockStagingStorage.AssertNotCalled(t, "SetLastPublishedBlockNumber", mock.Anything, mock.Anything)internal/orchestrator/committer.go (2)
307-329: Pre-commit publish flow: behavior matches PR intent; consider edge-case note
- Falls back to post-commit on cursor read error.
- Skips duplicate publish when
lastPublished >= highestBlock.- Updates cursor only after successful publish, which is correct.
Optional: document that in pre-commit failures, it’s possible to observe “published > committed” temporarily by design.
339-345: Post-commit publish retained; consider DRYing the publish callThe publish goroutine is duplicated across pre/post paths. Consider extracting a tiny helper (e.g.,
c.asyncPublish(blockData)) to keep logs and behavior consistent.
📜 Review details
Configuration used: CodeRabbit UI
Review profile: CHILL
Plan: Pro
📒 Files selected for processing (10)
cmd/root.go(2 hunks)configs/config.example.yml(1 hunks)configs/config.go(1 hunks)configs/test_config.yml(1 hunks)internal/orchestrator/committer.go(2 hunks)internal/orchestrator/committer_test.go(3 hunks)internal/storage/clickhouse.go(1 hunks)internal/storage/connector.go(1 hunks)internal/storage/postgres.go(1 hunks)test/mocks/MockIStagingStorage.go(1 hunks)
🧰 Additional context used
🧠 Learnings (1)
📚 Learning: 2025-06-04T11:28:38.057Z
Learnt from: iuwqyir
PR: thirdweb-dev/insight#216
File: internal/orchestrator/chain_tracker.go:45-53
Timestamp: 2025-06-04T11:28:38.057Z
Learning: Go's math/big.Int type has a Float64() method that returns (float64, bool), where the boolean indicates if the conversion was exact.
Applied to files:
internal/orchestrator/committer_test.go
🧬 Code Graph Analysis (1)
internal/orchestrator/committer.go (3)
internal/common/block.go (1)
Block(8-33)configs/config.go (1)
Cfg(208-208)internal/publisher/publisher.go (1)
Publisher(21-24)
🔇 Additional comments (9)
configs/test_config.yml (1)
67-67: LGTM: test config declarespublisher.modeThis keeps tests deterministic with explicit mode. Consider a separate test case toggling
pre-commitpath if not already covered.cmd/root.go (1)
254-254: LGTM: viper binding forpublisher.modeBinding is correct and matches the flag name and config path.
internal/storage/clickhouse.go (1)
1079-1096: Cursor query is already safe and needs no further hardening
- The ClickHouse
cursorstable is created as a ReplacingMergeTree ordered by(chain_id, cursor_type), so with theFINALmodifier it always returns at most one deduplicated row per key.- Calling
QueryRowon that query will yield either exactly one row orsql.ErrNoRows—addingLIMIT 1is therefore redundant.- Parameterization isn’t supported by the ClickHouse client here, and none of the interpolated values are user-supplied.
internal/storage/postgres.go (1)
366-374: Upsert looks correct and consistent with existing cursor handling
INSERT ... ON CONFLICTagainst(chain_id, cursor_type)is the right choice; passing numeric values as strings matches existing patterns.internal/orchestrator/committer_test.go (2)
5-5: Import oferrorsis correct and usedUsage in fallback test justifies the new import.
328-332: IncludingChainIdin test block data is necessaryCommit/publish code reads chain ID; this makes tests closer to real execution paths.
internal/orchestrator/committer.go (1)
298-304: Highest-block detection is correct and efficientSingle pass with big.Int comparison is clean; no redundant sorts here.
test/mocks/MockIStagingStorage.go (2)
75-132: Mock: GetLastPublishedBlockNumber additions look correctSignatures and generated helpers match the interface and test usage.
133-178: Mock: SetLastPublishedBlockNumber additions look correctCovers Run/Return/RunAndReturn; aligns with tests.
cmd/root.go
Outdated
| rootCmd.PersistentFlags().Bool("api-contractApiRequest-disableCompression", false, "Disable compression for contract API request") | ||
| rootCmd.PersistentFlags().Int("api-contractApiRequest-timeout", 10, "Timeout in seconds for contract API request") | ||
| rootCmd.PersistentFlags().Bool("publisher-enabled", false, "Toggle publisher") | ||
| rootCmd.PersistentFlags().String("publisher-mode", "post-commit", "Publisher mode: pre-commit or post-commit") |
There was a problem hiding this comment.
🛠️ Refactor suggestion
Validate --publisher-mode at startup to fail fast on invalid values
The help string documents allowed values, but invalid inputs will silently flow through as strings.
Add a validation step after config load (e.g., in initConfig() or inside configs.LoadConfig) to ensure publisher.mode ∈ {"pre-commit","post-commit"} and normalize case. Prefer returning an error (or exiting) if invalid to avoid unexpected runtime behavior.
🤖 Prompt for AI Agents
In cmd/root.go at line 138, the --publisher-mode flag accepts any string without
validation, which can cause silent errors. Add a validation step after loading
the configuration (such as in initConfig() or configs.LoadConfig) to check if
the publisher.mode value is either "pre-commit" or "post-commit"
(case-insensitive). Normalize the input to lowercase and if the value is
invalid, return an error or exit the program immediately to prevent unexpected
runtime behavior.
|
|
||
| type PublisherConfig struct { | ||
| Enabled bool `mapstructure:"enabled"` | ||
| Mode string `mapstructure:"mode"` |
There was a problem hiding this comment.
🛠️ Refactor suggestion
Validate and constrain publisher.mode to known values
Mode is a free-form string. Add typed constants and validate/normalize it during config load to prevent silent misconfiguration.
Example (outside this line range):
// Near other config-level constants
const (
PublisherModePreCommit = "pre-commit"
PublisherModePostCommit = "post-commit"
)
// In LoadConfig(), after viper.Unmarshal(&Cfg) and before returning
Cfg.Publisher.Mode = strings.TrimSpace(strings.ToLower(Cfg.Publisher.Mode))
if Cfg.Publisher.Mode == "" {
Cfg.Publisher.Mode = PublisherModePostCommit
}
switch Cfg.Publisher.Mode {
case PublisherModePreCommit, PublisherModePostCommit:
// ok
default:
return fmt.Errorf("invalid publisher.mode: %q (allowed: %q, %q)", Cfg.Publisher.Mode, PublisherModePreCommit, PublisherModePostCommit)
}Optional: expose these constants to other packages to avoid stringly-typed comparisons across the codebase.
🤖 Prompt for AI Agents
In configs/config.go at line 175, the Mode field is a free-form string without
validation, which can lead to silent misconfiguration. Define typed constants
for allowed modes (e.g., PublisherModePreCommit and PublisherModePostCommit)
near other config constants. In the LoadConfig function, after unmarshaling the
config, normalize the Mode value by trimming spaces and converting to lowercase,
set a default if empty, and validate it against the allowed constants, returning
an error if invalid. Optionally, export these constants for use in other
packages to avoid stringly-typed comparisons.
| func (c *ClickHouseConnector) SetLastPublishedBlockNumber(chainId *big.Int, blockNumber *big.Int) error { | ||
| query := fmt.Sprintf("INSERT INTO %s.cursors (chain_id, cursor_type, cursor_value) VALUES (%s, 'publish', '%s')", c.cfg.Database, chainId, blockNumber.String()) | ||
| return c.conn.Exec(context.Background(), query) | ||
| } |
There was a problem hiding this comment.
🛠️ Refactor suggestion
Use placeholders instead of string interpolation; fix %s with non-string
fmt.Sprintf(... %s, chainId, ...) relies on %s with a non-string. Prefer placeholders and arg binding (consistent with other methods).
-func (c *ClickHouseConnector) SetLastPublishedBlockNumber(chainId *big.Int, blockNumber *big.Int) error {
- query := fmt.Sprintf("INSERT INTO %s.cursors (chain_id, cursor_type, cursor_value) VALUES (%s, 'publish', '%s')", c.cfg.Database, chainId, blockNumber.String())
- return c.conn.Exec(context.Background(), query)
-}
+func (c *ClickHouseConnector) SetLastPublishedBlockNumber(chainId *big.Int, blockNumber *big.Int) error {
+ query := fmt.Sprintf("INSERT INTO %s.cursors (chain_id, cursor_type, cursor_value) VALUES (?, 'publish', ?)", c.cfg.Database)
+ return c.conn.Exec(context.Background(), query, chainId, blockNumber.String())
+}📝 Committable suggestion
‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.
| func (c *ClickHouseConnector) SetLastPublishedBlockNumber(chainId *big.Int, blockNumber *big.Int) error { | |
| query := fmt.Sprintf("INSERT INTO %s.cursors (chain_id, cursor_type, cursor_value) VALUES (%s, 'publish', '%s')", c.cfg.Database, chainId, blockNumber.String()) | |
| return c.conn.Exec(context.Background(), query) | |
| } | |
| func (c *ClickHouseConnector) SetLastPublishedBlockNumber(chainId *big.Int, blockNumber *big.Int) error { | |
| query := fmt.Sprintf( | |
| "INSERT INTO %s.cursors (chain_id, cursor_type, cursor_value) VALUES (?, 'publish', ?)", | |
| c.cfg.Database, | |
| ) | |
| return c.conn.Exec(context.Background(), query, chainId, blockNumber.String()) | |
| } |
🤖 Prompt for AI Agents
In internal/storage/clickhouse.go around lines 1098 to 1101, the SQL query uses
fmt.Sprintf with %s placeholders for non-string types like chainId, which is
incorrect and unsafe. Refactor the method to use parameter placeholders in the
query string and pass chainId and blockNumber as arguments to the Exec method,
ensuring proper argument binding and preventing SQL injection.
| GetLastPublishedBlockNumber(chainId *big.Int) (maxBlockNumber *big.Int, err error) | ||
| SetLastPublishedBlockNumber(chainId *big.Int, blockNumber *big.Int) error | ||
| DeleteOlderThan(chainId *big.Int, blockNumber *big.Int) error |
There was a problem hiding this comment.
💡 Verification agent
🧩 Analysis chain
Specify “not found” and concurrency semantics for publisher cursor methods
Ambiguity here can lead to bugs across drivers:
- Define whether “no cursor yet” returns (nil, nil) vs (0, nil). Recommend returning (nil, nil) and documenting it to avoid accidental use of zero as a real block height.
- Ensure
SetLastPublishedBlockNumberis idempotent and implemented as an upsert to avoid races when pre-commit publishing runs in a goroutine.
To confirm coverage and consistency, run:
🏁 Script executed:
#!/bin/bash
set -euo pipefail
echo "Search implementations of new cursor methods..."
rg -n "GetLastPublishedBlockNumber\(" -A 2
rg -n "SetLastPublishedBlockNumber\(" -A 2
echo "Check committer uses of publisher mode and cursor methods..."
rg -n "publisher\.mode|pre-commit|post-commit|GetLastPublishedBlockNumber|SetLastPublishedBlockNumber" -A 3 internalLength of output: 10576
🏁 Script executed:
#!/bin/bash
set -euo pipefail
echo "Locate DeleteOlderThan implementations in storage connectors..."
rg -n "func .*DeleteOlderThan" -A 10 internal/storageLength of output: 1746
Specify “not found” semantics and ensure idempotent upsert for publisher cursor methods
We need to be explicit about what “no cursor yet” means and make sure both drivers handle and document it uniformly, as well as guarantee that setting the cursor is safe under concurrent pre-commit runs.
– internal/storage/connector.go:
• Add a doc comment to GetLastPublishedBlockNumber stating that it returns (nil, nil) when no record exists.
• Note in the interface that SetLastPublishedBlockNumber must be idempotent (an upsert).
– internal/storage/postgres.go:
• Catch sql.ErrNoRows in GetLastPublishedBlockNumber, and return (nil, nil) instead of propagating the error.
• (Already in place) SetLastPublishedBlockNumber uses ON CONFLICT … DO UPDATE—no change needed here.
– internal/storage/clickhouse.go:
• In GetLastPublishedBlockNumber, detect when the query returns no rows (e.g. via rows.Next() or ErrNoRows) and return (nil, nil).
• Change SetLastPublishedBlockNumber to use a deduplicating insert (for example, a REPLACE INTO or a dedup‐capable MergeTree engine) so concurrent calls don’t create duplicate cursor entries; reading with FINAL alone is not sufficient to guarantee idempotence under race.
– Tests:
• Update any mocks to expect (nil, nil) instead of an error for the “no cursor” case.
• Add coverage for concurrent pre-commit calls to SetLastPublishedBlockNumber and verify that only one logical update occurs.
🤖 Prompt for AI Agents
In internal/storage/connector.go around lines 86 to 88, add a doc comment to
GetLastPublishedBlockNumber specifying it returns (nil, nil) when no record
exists, and note that SetLastPublishedBlockNumber must be idempotent as an
upsert. In internal/storage/postgres.go, modify GetLastPublishedBlockNumber to
catch sql.ErrNoRows and return (nil, nil) instead of an error. In
internal/storage/clickhouse.go, update GetLastPublishedBlockNumber to detect no
rows returned and return (nil, nil), and change SetLastPublishedBlockNumber to
use a deduplicating insert method like REPLACE INTO or a dedup-capable MergeTree
engine to ensure idempotence under concurrent calls. Finally, update tests and
mocks to expect (nil, nil) for no cursor cases and add tests verifying
idempotent behavior of concurrent SetLastPublishedBlockNumber calls.
Co-authored-by: coderabbitai[bot] <136622811+coderabbitai[bot]@users.noreply.github.com> Signed-off-by: Jake Loo <[email protected]>
Summary
publisher-modeshouldPostCommitPublishflagTesting
go test ./...https://chatgpt.com/codex/tasks/task_b_6896dc37689c8332b734b4405875abf1
Summary by CodeRabbit
New Features
Tests
Documentation