Merged
Conversation
|
The latest updates on your projects. Learn more about Vercel for GitHub.
|
Contributor
|
Note Reviews pausedIt looks like this branch is under active development. To avoid overwhelming you with review comments due to an influx of new commits, CodeRabbit has automatically paused this review. You can configure this behavior by changing the Use the following commands to manage reviews:
Use the checkboxes below for quick actions:
📝 WalkthroughWalkthroughDecouples buffering from the ClickHouse client by introducing generic Changes
Sequence Diagram(s)sequenceDiagram
participant Client as ClickHouse.Client
participant Buffer as batch.BatchProcessor[T]
participant Producer as Service/Handler
participant DB as ClickHouse Table
rect rgba(200,200,255,0.5)
Producer->>Buffer: Buffer(event T)
end
rect rgba(200,255,200,0.5)
Buffer->>Buffer: accumulate, async consumers
Buffer->>Client: flush(rows []T)
Client->>DB: async INSERT into table
DB-->>Client: OK / error
Client-->>Buffer: success/error
end
Estimated code review effort🎯 4 (Complex) | ⏱️ ~60 minutes Possibly related PRs
🚥 Pre-merge checks | ✅ 2 | ❌ 1❌ Failed checks (1 warning)
✅ Passed checks (2 passed)
✏️ Tip: You can configure your own custom pre-merge checks in the settings. ✨ Finishing Touches🧪 Generate unit tests (beta)
Comment |
Flo4604
requested review from
MichaelUnkey,
chronark,
imeyer,
mcstepp,
ogzhanolguncu and
perkinsjr
as code owners
Flo4604
force-pushed
the
03-27-chore_rework_clickhouse_package
branch
from
34d15aa to
008d03b
Compare
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
Contributor
There was a problem hiding this comment.
Actionable comments posted: 6
Caution
Some comments are outside the diff and can’t be posted inline due to platform limitations.
⚠️ Outside diff range comments (3)
cmd/dev/seed/sentinel.go (1)
239-250:⚠️ Potential issue | 🟠 MajorSeed one sentinel region only.
Line 249 still samples from a cross-region pool. Sentinel is intra-region-only, so this writes impossible telemetry for one deployment and skews the charts this seed command is meant to validate. Drive
Regionfrom the deployment/local region, or pin one region per run.Based on learnings, "In the Unkeyed architecture, sentinel components are deployed intra-region only ... ensure it remains constrained to the local region."
🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@cmd/dev/seed/sentinel.go` around lines 239 - 250, The Region field of the seeded schema.SentinelRequest is incorrectly using weightedSelectString(regions) (cross-region sampling) which produces impossible cross-region sentinel telemetry; update the seeding logic in the code that calls s.sentinelRequests.Buffer (where schema.SentinelRequest is constructed) to derive Region from the current deployment/local region (e.g., use deployment.Region or a single pinned region for the run) instead of weightedSelectString(regions) so that Region is constrained intra-region-only for the seeded telemetry.svc/ctrl/worker/run.go (1)
155-178:⚠️ Potential issue | 🟠 MajorDrain these processors on shutdown.
Both buffers are async. Without
Close(), the tail of build steps/logs is dropped on graceful stop, and the processor goroutines outlive the service. Register both closes before closingchClient.Minimal fix
} else { ch = chClient + r.Defer(chClient.Close) buildSteps = clickhouse.NewBuffer[schema.BuildStepV1](chClient, "default.build_steps_v1", clickhouse.BufferConfig{ Name: "build_steps", @@ -168,6 +169,8 @@ func Run(ctx context.Context, cfg Config) error { FlushInterval: 2 * time.Second, Consumers: 1, }) + r.Defer(func() error { buildStepLogs.Close(); return nil }) + r.Defer(func() error { buildSteps.Close(); return nil }) } }🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@svc/ctrl/worker/run.go` around lines 155 - 178, The async ClickHouse buffers buildSteps and buildStepLogs must be closed on shutdown to flush remaining items and stop their goroutines before closing the underlying client (chClient/ch); update the shutdown/cleanup path in run.go to call buildSteps.Close() and buildStepLogs.Close() (if non-nil) prior to closing chClient (and set or check ch) so buffers flush and their goroutines terminate cleanly; ensure nil checks and error handling around Close() calls and register these closes in the same shutdown sequence where chClient is closed.svc/ctrl/worker/deploy/service.go (1)
111-127:⚠️ Potential issue | 🟠 MajorDefault nil processors to noop in
New.
build.gonow buffers through these fields. If a caller misses the new config,Newstill returns a workflow that nil-derefs on the first build event. Normalize nil tobatch.NewNoop[...]()here.Minimal fix
func New(cfg Config) *Workflow { + buildSteps := cfg.BuildSteps + if buildSteps == nil { + buildSteps = batch.NewNoop[schema.BuildStepV1]() + } + buildStepLogs := cfg.BuildStepLogs + if buildStepLogs == nil { + buildStepLogs = batch.NewNoop[schema.BuildStepLogV1]() + } + return &Workflow{ UnimplementedDeployServiceServer: hydrav1.UnimplementedDeployServiceServer{}, db: cfg.DB, @@ -121,8 +129,8 @@ func New(cfg Config) *Workflow { registryConfig: cfg.RegistryConfig, buildPlatform: cfg.BuildPlatform, clickhouse: cfg.Clickhouse, - buildSteps: cfg.BuildSteps, - buildStepLogs: cfg.BuildStepLogs, + buildSteps: buildSteps, + buildStepLogs: buildStepLogs, allowUnauthenticatedDeployments: cfg.AllowUnauthenticatedDeployments, dashboardURL: cfg.DashboardURL, }As per coding guidelines, "Make illegal states unrepresentable by modeling domain with ADTs/discriminated unions and parsing inputs at boundaries into typed structures".
🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@svc/ctrl/worker/deploy/service.go` around lines 111 - 127, The New constructor returns a Workflow with processor fields that may be nil and later cause nil-derefs during builds; in New, check processor-like fields (at minimum buildSteps and buildStepLogs) and if they are nil assign them to batch.NewNoop[...]() of the correct generic type so the Workflow always has noop processors; update the initialization in New to normalize nil -> batch.NewNoop for those fields (reference the New function and Workflow struct and the buildSteps / buildStepLogs fields) so callers who omit these configs won’t crash.
🧹 Nitpick comments (4)
svc/sentinel/config.go (1)
15-28: Config fields for batch processor tuning look good.Reasonable defaults (BatchSize=5000, BufferSize=10000, Consumers=1) with min=1 validation.
One consideration: if
BatchSize > BufferSize, batches would never fill completely before blocking. You may want to document or validate thatBufferSize >= BatchSizefor optimal behavior.🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@svc/sentinel/config.go` around lines 15 - 28, The config allows BatchSize and BufferSize to be set independently (BatchSize, BufferSize, Consumers) which can lead to BufferSize < BatchSize causing batches to never fill; add validation after config parsing (or in the struct's validation hook) to ensure BufferSize >= BatchSize and return a clear config error if not (or automatically set BufferSize = BatchSize with a warning), and update any docs/comments to state the invariant "BufferSize must be >= BatchSize" for optimal behavior.pkg/clickhouse/billable_ratelimits.go (1)
19-46: Useerrors.Isfor the no-row case.Line 41 matches the error string. That is brittle; wrapped
sql.ErrNoRowswill miss and turn an empty month into a fault. Switch this helper — and the sibling billable helpers touched in this PR — toerrors.Is(err, sql.ErrNoRows).♻️ Minimal diff
import ( "context" + "database/sql" + "errors" "github.com/unkeyed/unkey/pkg/fault" ) @@ - if err != nil && err.Error() == "sql: no rows in result set" { + if errors.Is(err, sql.ErrNoRows) { return 0, nil }🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@pkg/clickhouse/billable_ratelimits.go` around lines 19 - 46, The no-rows check in GetBillableRatelimits is brittle because it compares the error string; update the check to use errors.Is(err, sql.ErrNoRows) instead (importing "errors" and "database/sql" if not already), and apply the same change to the sibling billable helper functions touched in this PR so they all detect a no-row result via errors.Is(err, sql.ErrNoRows) and return (0, nil) in that case while preserving the existing fault.Wrap handling for other errors.cmd/dev/seed/sentinel.go (1)
43-56: Verify the buffer ownschbefore dropping it.After Line 50 the file loses the only
Clientreference, and Line 276 only closes the batcher. IfBatchProcessor.Close()does not also close the underlying client, this command leaks the ClickHouse connection on success and on every early return.Also applies to: 79-84, 275-276
🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@cmd/dev/seed/sentinel.go` around lines 43 - 56, The ClickHouse client stored in ch is potentially leaked because only the batcher (sentinelRequests) is closed later; inspect clickhouse.NewBuffer and BatchProcessor.Close to determine ownership, and then ensure the client is closed if the buffer does not own it: either (A) update the code that constructs the buffer (clickhouse.NewBuffer) to pass an ownership flag or use a constructor that transfers ownership, or (B) explicitly close ch after closing sentinelRequests (call ch.Close() or ch.Shutdown()) and add a defer or explicit closes on all early returns so ch is always closed; refer to the ch variable from clickhouse.New and the sentinelRequests instance created by clickhouse.NewBuffer (and BatchProcessor.Close) when adding the client close.cmd/dev/seed/verifications.go (1)
52-67: Move buffer/client cleanup toseedVerifications.
Close()only runs on the happy path insidegenerateVerifications, andchnever closes. Deferring both right after construction makes all exit paths drain/cleanup consistently.Minimal fix
ch, err := clickhouse.New(clickhouse.Config{ URL: cmd.String("clickhouse-url"), }) if err != nil { return fmt.Errorf("failed to connect to ClickHouse: %w", err) } + defer ch.Close() keyVerifications := clickhouse.NewBuffer[schema.KeyVerification](ch, "default.key_verifications_raw_v2", clickhouse.BufferConfig{ Name: "seed-key-verifications", @@ -64,6 +65,7 @@ func seedVerifications(ctx context.Context, cmd *cli.Command) error { FlushInterval: 5 * time.Second, Consumers: 2, }) + defer keyVerifications.Close() // Create key service for proper key generation keyService, err := keys.New(keys.Config{ @@ -497,7 +499,5 @@ func (s *Seeder) generateVerifications(_ context.Context, workspaceID string, ke log.Printf(" Buffered all %d verifications, waiting for flush...", s.numVerifications) - - s.keyVerifications.Close() log.Printf(" All verifications sent to ClickHouse") return nilAlso applies to: 500-500
🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@cmd/dev/seed/verifications.go` around lines 52 - 67, The buffer/client cleanup (ch and the keyVerifications buffer) must be deferred immediately after construction so they run on all exit paths; move the defer calls for ch.Close() and keyVerifications.Close() out of generateVerifications and place them right after clickhouse.New(...) and clickhouse.NewBuffer(...) respectively (reference symbols: clickhouse.New, clickhouse.NewBuffer, ch, keyVerifications, seedVerifications, generateVerifications) to ensure both the ClickHouse client and the buffer are always drained/closed even on early returns or errors.
🤖 Prompt for all review comments with AI agents
Verify each finding against the current code and only fix it if needed.
Inline comments:
In `@internal/services/keys/service.go`:
- Around line 44-54: The New constructor currently allows keyVerifications to be
nil which leads to a nil-deref on first use; update New (the New function
returning *service) to default the keyVerifications field to
batch.NewNoop[schema.KeyVerification]() when config.KeyVerifications is nil so
the service.keyVerifications is always non-nil and safe to call.
In `@pkg/clickhouse/buffer.go`:
- Around line 46-58: The current NewBuffer helper hardcodes Drop: true and
swallows flush failures (it only logs via logger.Error in the Flush closure),
which is too lossy for non-telemetry flows; update the API so callers can choose
strict vs best-effort behavior: either add parameters/options to NewBuffer
(e.g., drop bool and onError func(error, context, table, rows)) or split into
two constructors (e.g., NewBestEffortBuffer and NewStrictBuffer). Wire the
chosen behavior into the batch.Config returned (set Drop from the
option/constructor and replace the inline Flush closure to call the configured
onError handler or propagate/report errors accordingly instead of only logging
via logger.Error). Ensure references: NewBuffer, batch.Config[T], the Drop
field, and the Flush closure that calls flush(c, ctx, table, rows) are updated
to honor the new option/constructor semantics.
In `@pkg/zen/middleware_metrics.go`:
- Line 8: The package import "github.com/unkeyed/unkey/pkg/batch" in
pkg/zen/middleware_metrics.go is not declared in the owning target's BUILD deps;
update the BUILD file for the pkg/zen package to add the
//pkg/batch:go_default_library (or the correct pkg/batch target) to the target's
deps so the import resolves and CI strict-deps passes.
In `@svc/api/internal/testutil/http.go`:
- Around line 123-139: The test buffers keyVerifications and ratelimitsfer
created via clickhouse.NewBuffer (using clickhouse.BufferConfig) are configured
with BatchSize/BufferSize=50_000 and FlushInterval=5s which causes stale reads
in tests; either reduce BatchSize/BufferSize to a very small number (e.g., 1 or
10) and set FlushInterval to a near-immediate duration (e.g., 10-100ms) in the
BufferConfig for keyVerifications and ratelimitsfer, or add an explicit
synchronous flush/Close/Flush call on these buffer objects
(keyVerifications.Flush or keyVerifications.Close and ratelimitsfer.Flush/Close)
immediately before assertions to ensure test data is persisted to ClickHouse.
In `@svc/api/run.go`:
- Around line 162-166: The defer registration order is reversed so
chClient.Close runs before buffer Close calls, causing buffer flushes to occur
after the connection is closed; fix by registering chClient.Close with r.Defer
before registering the buffer Close defers so that LIFO execution closes buffers
first and the client last—update the r.Defer call order involving
chClient.Close, apiRequests.Close, keyVerifications.Close, and ratelimits.Close
accordingly.
In `@svc/sentinel/run.go`:
- Around line 141-144: The defer registrations are in the wrong order: r.Defer
currently registers sentinelRequests.Close and keyVerifications.Close before
chClient.Close which causes chClient to be closed first and can break final
flush; fix by registering r.Defer(chClient.Close) before registering
r.Defer(func() error { sentinelRequests.Close(); return nil }) and
r.Defer(func() error { keyVerifications.Close(); return nil }) so closures
happen LIFO with buffers closed after the channel; update the r.Defer call order
for chClient.Close, sentinelRequests.Close, and keyVerifications.Close
accordingly.
---
Outside diff comments:
In `@cmd/dev/seed/sentinel.go`:
- Around line 239-250: The Region field of the seeded schema.SentinelRequest is
incorrectly using weightedSelectString(regions) (cross-region sampling) which
produces impossible cross-region sentinel telemetry; update the seeding logic in
the code that calls s.sentinelRequests.Buffer (where schema.SentinelRequest is
constructed) to derive Region from the current deployment/local region (e.g.,
use deployment.Region or a single pinned region for the run) instead of
weightedSelectString(regions) so that Region is constrained intra-region-only
for the seeded telemetry.
In `@svc/ctrl/worker/deploy/service.go`:
- Around line 111-127: The New constructor returns a Workflow with processor
fields that may be nil and later cause nil-derefs during builds; in New, check
processor-like fields (at minimum buildSteps and buildStepLogs) and if they are
nil assign them to batch.NewNoop[...]() of the correct generic type so the
Workflow always has noop processors; update the initialization in New to
normalize nil -> batch.NewNoop for those fields (reference the New function and
Workflow struct and the buildSteps / buildStepLogs fields) so callers who omit
these configs won’t crash.
In `@svc/ctrl/worker/run.go`:
- Around line 155-178: The async ClickHouse buffers buildSteps and buildStepLogs
must be closed on shutdown to flush remaining items and stop their goroutines
before closing the underlying client (chClient/ch); update the shutdown/cleanup
path in run.go to call buildSteps.Close() and buildStepLogs.Close() (if non-nil)
prior to closing chClient (and set or check ch) so buffers flush and their
goroutines terminate cleanly; ensure nil checks and error handling around
Close() calls and register these closes in the same shutdown sequence where
chClient is closed.
---
Nitpick comments:
In `@cmd/dev/seed/sentinel.go`:
- Around line 43-56: The ClickHouse client stored in ch is potentially leaked
because only the batcher (sentinelRequests) is closed later; inspect
clickhouse.NewBuffer and BatchProcessor.Close to determine ownership, and then
ensure the client is closed if the buffer does not own it: either (A) update the
code that constructs the buffer (clickhouse.NewBuffer) to pass an ownership flag
or use a constructor that transfers ownership, or (B) explicitly close ch after
closing sentinelRequests (call ch.Close() or ch.Shutdown()) and add a defer or
explicit closes on all early returns so ch is always closed; refer to the ch
variable from clickhouse.New and the sentinelRequests instance created by
clickhouse.NewBuffer (and BatchProcessor.Close) when adding the client close.
In `@cmd/dev/seed/verifications.go`:
- Around line 52-67: The buffer/client cleanup (ch and the keyVerifications
buffer) must be deferred immediately after construction so they run on all exit
paths; move the defer calls for ch.Close() and keyVerifications.Close() out of
generateVerifications and place them right after clickhouse.New(...) and
clickhouse.NewBuffer(...) respectively (reference symbols: clickhouse.New,
clickhouse.NewBuffer, ch, keyVerifications, seedVerifications,
generateVerifications) to ensure both the ClickHouse client and the buffer are
always drained/closed even on early returns or errors.
In `@pkg/clickhouse/billable_ratelimits.go`:
- Around line 19-46: The no-rows check in GetBillableRatelimits is brittle
because it compares the error string; update the check to use errors.Is(err,
sql.ErrNoRows) instead (importing "errors" and "database/sql" if not already),
and apply the same change to the sibling billable helper functions touched in
this PR so they all detect a no-row result via errors.Is(err, sql.ErrNoRows) and
return (0, nil) in that case while preserving the existing fault.Wrap handling
for other errors.
In `@svc/sentinel/config.go`:
- Around line 15-28: The config allows BatchSize and BufferSize to be set
independently (BatchSize, BufferSize, Consumers) which can lead to BufferSize <
BatchSize causing batches to never fill; add validation after config parsing (or
in the struct's validation hook) to ensure BufferSize >= BatchSize and return a
clear config error if not (or automatically set BufferSize = BatchSize with a
warning), and update any docs/comments to state the invariant "BufferSize must
be >= BatchSize" for optimal behavior.
🪄 Autofix (Beta)
Fix all unresolved CodeRabbit comments on this PR:
- Push a commit to this branch (recommended)
- Create a new PR with the fixes
ℹ️ Review info
⚙️ Run configuration
Configuration used: Repository UI
Review profile: CHILL
Plan: Pro
Run ID: 63f5784b-20bc-4ff1-8a3f-6988560944f7
📒 Files selected for processing (58)
cmd/dev/seed/local.gocmd/dev/seed/sentinel.gocmd/dev/seed/verifications.gointernal/services/keys/get.gointernal/services/keys/service.gointernal/services/keys/verifier.gopkg/batch/noop.gopkg/clickhouse/billable_ratelimits.gopkg/clickhouse/billable_usage.gopkg/clickhouse/billable_verifications.gopkg/clickhouse/buffer.gopkg/clickhouse/client.gopkg/clickhouse/deployment_request_count.gopkg/clickhouse/flush.gopkg/clickhouse/interface.gopkg/clickhouse/key_last_used.gopkg/clickhouse/noop.gopkg/clickhouse/user.gopkg/zen/middleware_metrics.gosvc/api/internal/testutil/http.gosvc/api/routes/register.gosvc/api/routes/services.gosvc/api/routes/v2_analytics_get_verifications/200_test.gosvc/api/routes/v2_analytics_get_verifications/400_test.gosvc/api/routes/v2_analytics_get_verifications/401_test.gosvc/api/routes/v2_analytics_get_verifications/403_test.gosvc/api/routes/v2_analytics_get_verifications/404_test.gosvc/api/routes/v2_analytics_get_verifications/412_test.gosvc/api/routes/v2_analytics_get_verifications/422_test.gosvc/api/routes/v2_analytics_get_verifications/429_test.gosvc/api/routes/v2_analytics_get_verifications/503_test.gosvc/api/routes/v2_analytics_get_verifications/handler.gosvc/api/routes/v2_keys_verify_key/200_test.gosvc/api/routes/v2_keys_verify_key/400_test.gosvc/api/routes/v2_keys_verify_key/401_test.gosvc/api/routes/v2_keys_verify_key/403_test.gosvc/api/routes/v2_keys_verify_key/404_test.gosvc/api/routes/v2_keys_verify_key/412_test.gosvc/api/routes/v2_keys_verify_key/handler.gosvc/api/routes/v2_keys_verify_key/migration_test.gosvc/api/routes/v2_keys_verify_key/multilimit_test.gosvc/api/routes/v2_keys_verify_key/multiple_ratelimits_overcommit_test.gosvc/api/routes/v2_keys_verify_key/ratelimit_response_test.gosvc/api/routes/v2_keys_verify_key/resend_demo_test.gosvc/api/routes/v2_ratelimit_limit/200_test.gosvc/api/routes/v2_ratelimit_limit/accuracy_test.gosvc/api/routes/v2_ratelimit_limit/handler.gosvc/api/routes/v2_ratelimit_multi_limit/200_test.gosvc/api/routes/v2_ratelimit_multi_limit/handler.gosvc/api/run.gosvc/ctrl/worker/deploy/build.gosvc/ctrl/worker/deploy/service.gosvc/ctrl/worker/run.gosvc/sentinel/config.gosvc/sentinel/middleware/logging.gosvc/sentinel/routes/register.gosvc/sentinel/routes/services.gosvc/sentinel/run.go
💤 Files with no reviewable changes (8)
- svc/api/routes/v2_analytics_get_verifications/412_test.go
- svc/api/routes/v2_analytics_get_verifications/503_test.go
- svc/api/routes/v2_analytics_get_verifications/404_test.go
- svc/api/routes/v2_analytics_get_verifications/403_test.go
- svc/api/routes/v2_analytics_get_verifications/400_test.go
- pkg/clickhouse/interface.go
- svc/api/routes/v2_analytics_get_verifications/handler.go
- svc/api/routes/v2_analytics_get_verifications/401_test.go
Flo4604
force-pushed
the
03-27-chore_rework_clickhouse_package
branch
from
008d03b to
3489a85
Compare
Contributor
There was a problem hiding this comment.
Actionable comments posted: 2
Caution
Some comments are outside the diff and can’t be posted inline due to platform limitations.
⚠️ Outside diff range comments (3)
cmd/dev/seed/sentinel.go (1)
42-56:⚠️ Potential issue | 🟡 MinorMissing ClickHouse client close on shutdown.
The ClickHouse client
chis created but never closed. The batch processor'sClose()only drains the buffer - it doesn't close the underlying connection.🐛 Add deferred close for ClickHouse client
ch, err := clickhouse.New(clickhouse.Config{ URL: cmd.String("clickhouse-url"), }) if err != nil { return fmt.Errorf("failed to connect to ClickHouse: %w", err) } + defer ch.Close() sentinelRequests := clickhouse.NewBuffer[schema.SentinelRequest](ch, "default.sentinel_requests_raw_v1", clickhouse.BufferConfig{🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@cmd/dev/seed/sentinel.go` around lines 42 - 56, The ClickHouse client created via clickhouse.New (variable ch) is never closed; add a deferred or explicit close for ch when seeding to ensure the underlying connection is released (the Buffer.Close on sentinelRequests only drains buffers). Locate the clickhouse.New(...) call that returns ch and err and add a defer or final ch.Close()/ch.CloseContext(...) (matching the client's API) after successful creation, ensuring it runs after sentinelRequests is used and after any errors are handled.pkg/clickhouse/client.go (2)
186-195:⚠️ Potential issue | 🟠 MajorThis shutdown contract is already broken once.
Client.Closeno longer drains buffers, butsvc/ctrl/worker/run.go:150-175still createsbuildStepsandbuildStepLogswithout anyClose().svc/api/run.go:163-166shows the needed ordering. As-is, shutdown drops pending rows. Please make buffer closure enforceable here, or fix the missing closes before merge.🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@pkg/clickhouse/client.go` around lines 186 - 195, Client.Close currently just closes c.conn, which drops pending buffer rows; either enforce buffer closure inside Client or require callers to close buffers before calling Client.Close. Update the ClickHouse client implementation (functions NewBuffer and Client.Close) so that NewBuffer registers created buffers with the Client (e.g., a map/slice of active buffers) and Client.Close iterates and Close()s each registered buffer (unregistering as they close), returning any aggregated error; alternatively, make Client.Close detect active buffers and return an error if any are still open so callers must Close buffers first. Then update callers creating buffers (notably the buildSteps and buildStepLogs usages in svc/ctrl/worker/run.go and svc/api/run.go) to explicitly Close() those buffers before calling Client.Close to preserve shutdown ordering.
52-89:⚠️ Potential issue | 🟠 MajorClose failed connections to avoid resource leaks.
If
Pingexhausts retries,Newreturns afterch.Opensucceeds, leaving the ClickHouse connection open indefinitely. The resource is never garbage collected and the connection pool persists for the process lifetime.🩹 Proposed fix
func New(config Config) (*Client, error) { opts, err := ch.ParseDSN(config.URL) if err != nil { return nil, fault.Wrap(err, fault.Internal("parsing clickhouse DSN failed")) } @@ conn, err := ch.Open(opts) if err != nil { return nil, fault.Wrap(err, fault.Internal("opening clickhouse failed")) } + closeOnError := true + defer func() { + if closeOnError { + _ = conn.Close() + } + }() err = retry.New( @@ if err != nil { return nil, fault.Wrap(err, fault.Internal("pinging clickhouse failed")) } @@ - return c, nil + closeOnError = false + return c, nil }🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@pkg/clickhouse/client.go` around lines 52 - 89, New currently opens a ClickHouse connection with ch.Open(opts) and if subsequent ping/retry fails it returns an error without closing conn, leaking the connection; update New so that after ch.Open(opts) succeeds you ensure conn.Close() is called on all error paths (e.g. if retry.Do against conn.Ping fails) — either by deferring a conditional close (defer func(){ if !clientCreated { _ = conn.Close() } }()) or by explicitly calling conn.Close() before each return that occurs after ch.Open; reference the New function, the conn variable, ch.Open, retry.Do (conn.Ping) and ensure the successful return path still returns the open connection without closing it.
♻️ Duplicate comments (2)
svc/api/run.go (1)
162-166:⚠️ Potential issue | 🔴 CriticalDefer order still inverted - buffers will fail to flush on shutdown.
The past review flagged this issue and marked it "addressed", but the current code still has the same problematic order. With LIFO execution,
chClient.Close()(registered last) executes first, closing the connection before the buffer processors can flush.🐛 Fix: register client defer first so it closes last
- // Close buffers before connection (LIFO) - r.Defer(func() error { apiRequests.Close(); return nil }) - r.Defer(func() error { keyVerifications.Close(); return nil }) - r.Defer(func() error { ratelimits.Close(); return nil }) - r.Defer(chClient.Close) + // Close connection last (LIFO: first registered closes last) + r.Defer(chClient.Close) + r.Defer(func() error { ratelimits.Close(); return nil }) + r.Defer(func() error { keyVerifications.Close(); return nil }) + r.Defer(func() error { apiRequests.Close(); return nil })🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@svc/api/run.go` around lines 162 - 166, The defer registration order is inverted: because r.Defer runs LIFO, chClient.Close is currently registered last and will run first (closing the connection before buffers flush). Fix by registering the client close before the buffer closes so the client is closed last — i.e., call r.Defer(chClient.Close) first, then r.Defer for apiRequests.Close, keyVerifications.Close, and ratelimits.Close (or otherwise ensure chClient.Close is registered earliest) inside the same function where r.Defer is used.svc/sentinel/run.go (1)
141-144:⚠️ Potential issue | 🔴 CriticalDefer order inverted - buffers will fail to flush on shutdown.
Same issue as flagged in past review. With LIFO execution,
chClient.Close()(registered last) executes first, closing the connection before buffer processors can flush their remaining data.🐛 Minimal fix: reverse registration order
// Close buffers before connection (LIFO) - r.Defer(func() error { sentinelRequests.Close(); return nil }) - r.Defer(func() error { keyVerifications.Close(); return nil }) - r.Defer(chClient.Close) + r.Defer(chClient.Close) + r.Defer(func() error { keyVerifications.Close(); return nil }) + r.Defer(func() error { sentinelRequests.Close(); return nil })🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@svc/sentinel/run.go` around lines 141 - 144, The defer registration order is inverted: because r.Defer runs handlers LIFO, chClient.Close is currently registered last and thus executes first, closing the connection before buffer processors flush. Fix by registering chClient.Close first, then register keyVerifications.Close and sentinelRequests.Close (so they execute before chClient.Close); reference the r.Defer calls and the sentinelRequests.Close, keyVerifications.Close, and chClient.Close symbols when making the change.
🧹 Nitpick comments (5)
svc/api/routes/v2_analytics_get_verifications/200_test.go (1)
61-62: Consider usingrequire.EventuallyWithTfor consistency.Tests at lines 137 and 217 use the polling pattern which is more robust. The
time.Sleep(5 * time.Second)matches the flush interval exactly (per harness config), which could be flaky. Pre-existing pattern, but worth aligning for reliability.💡 Example refactor
- // Wait for buffered data to be available - time.Sleep(5 * time.Second) - - res := testutil.CallRoute[Request, Response](h, route, headers, req) - t.Logf("Status: %d, RawBody: %s", res.Status, res.RawBody) - require.Equal(t, 200, res.Status) - require.NotNil(t, res.Body) - require.Len(t, res.Body.Data, 1) + require.EventuallyWithT(t, func(c *assert.CollectT) { + res := testutil.CallRoute[Request, Response](h, route, headers, req) + require.Equal(c, 200, res.Status) + require.NotNil(c, res.Body) + require.Len(c, res.Body.Data, 1) + }, 30*time.Second, time.Second)🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@svc/api/routes/v2_analytics_get_verifications/200_test.go` around lines 61 - 62, Replace the brittle time.Sleep(5 * time.Second) with the same polling pattern used elsewhere in this test file: use require.EventuallyWithT to repeatedly check for the expected buffered/flush condition until it becomes true (e.g. a function that queries the verification buffer/count used by the other checks in this file) with a reasonable timeout and interval (timeout > 5s, interval ~100–250ms). Locate the sleep call in the test and swap it for require.EventuallyWithT(t, func(t assert.TestingT) bool { /* return true when buffer/flush visible */ }, timeout, interval) so the test aligns with the polling used in the other assertions and avoids flakiness.pkg/buffer/buffer_test.go (1)
145-146: Userequire.Equalin touched tests.New assertions should follow repo test style; using
asserthere keeps the old pattern alive in newly touched code.As per coding guidelines,
**/*.go: Usetestify/requirefor assertions in tests.🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@pkg/buffer/buffer_test.go` around lines 145 - 146, Replace the testify assert usage with require in this test: change assert.Equal(t, event, *received, ...) to require.Equal(t, event, *received) and update the test imports to use "github.com/stretchr/testify/require" instead of or in addition to assert; locate the comparison around the channel receive (variable received from b.c) in buffer_test.go and remove the custom message parameter so it uses require's signature with t, expected, actual.cmd/dev/seed/local.go (1)
44-52: Use the noop verifier buffer here, notnil.This is a ClickHouse-disabled path. Passing
nilkeepskeys.Configin a partially invalid state and can turn into a panic later if key-verification buffering gets exercised. Use the new noop batch processor instead.As per coding guidelines,
**/*.go: Make illegal states unrepresentable by modeling domain with ADTs/discriminated unions and parsing inputs at boundaries into typed structures.🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@cmd/dev/seed/local.go` around lines 44 - 52, The KeyVerifications field in keys.Config is being set to nil which leaves keys.Config in an invalid state and can panic later; replace the nil with the package's provided noop batch verifier so the ClickHouse-disabled path remains safe—update the keys.New call to set KeyVerifications to the noop batch processor (e.g., the provided keys.NoopBatchProcessor()/keys.NoopVerifier factory) instead of nil and ensure any related UsageLimiter/KeyCache/QuotaCache fields follow the same non-nil pattern if applicable.pkg/batch/process.go (1)
124-128: Consider pre-allocating with BatchSize hint.Starting with
make([]T, 0)means the slice will grow through multiple allocations until reaching steady state. For high-throughput buffers, considermake([]T, 0, bp.config.BatchSize)to pre-allocate capacity upfront while still reusing viabatch[:0].♻️ Optional: pre-allocate capacity
- batch := make([]T, 0) + batch := make([]T, 0, bp.config.BatchSize)🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@pkg/batch/process.go` around lines 124 - 128, Replace the zero-capacity slice allocation with a capacity hint to avoid repeated growth: change the `batch := make([]T, 0)` allocation to reserve `bp.config.BatchSize` capacity (e.g., make([]T, 0, bp.config.BatchSize)) so `batch` starts with sufficient capacity and can still be reused via `batch[:0]`; update any related comments referencing steady-state reuse if present.cmd/dev/seed/sentinel.go (1)
273-279: Shutdown ordering: close buffer before connection.The buffer must be closed before the ClickHouse client to ensure final flush completes. With the suggested
defer ch.Close()above, ensure the buffer close happens first (currently it does, but explicitly documenting intent helps).♻️ Explicit ordering comment
log.Printf(" Buffered all %d requests, waiting for flush...") - // Flush by closing the buffer + // Flush by closing the buffer (must happen before ch.Close() deferred above) s.sentinelRequests.Close()🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@cmd/dev/seed/sentinel.go` around lines 273 - 279, Ensure the sentinel request buffer is closed before the ClickHouse client is closed so the final flush can complete: explicitly call s.sentinelRequests.Close() prior to closing the ClickHouse client (ch.Close()) or, if you used defer ch.Close(), remove or delay that defer so it runs after s.sentinelRequests.Close(); update the code around the log lines and the functions that manage the lifetime (e.g., s.sentinelRequests.Close() and ch.Close()) to guarantee the buffer close happens first and add a short comment documenting this ordering intent.
🤖 Prompt for all review comments with AI agents
Verify each finding against the current code and only fix it if needed.
Inline comments:
In `@internal/services/usagelimiter/redis.go`:
- Around line 359-360: The select branch reading from s.replayBuffer.Consume()
dereferences ptr without checking for a closed-channel receive; change the
receive to use the comma-ok form (e.g., ptr, ok := <-s.replayBuffer.Consume())
or explicitly check ptr for nil and only call s.syncWithDB(...) when the pointer
is non-nil, so that syncWithDB is never called with a nil dereference during
shutdown in the replay loop that handles s.replayBuffer and replayRequests.
In `@svc/ctrl/worker/run.go`:
- Around line 162-178: The buffers buildSteps and buildStepLogs and the
ClickHouse client chClient are not closed on shutdown; add deferred shutdowns
immediately after creation to avoid data loss: defer buildStepLogs.Close(...),
defer buildSteps.Close(...), then defer chClient.Close(...) in LIFO order (close
logs first, then steps, then client), and propagate or log any returned errors
from those Close calls; reference the buffer variables buildSteps, buildStepLogs
and the chClient instance created via clickhouse.NewBuffer and chClient.
---
Outside diff comments:
In `@cmd/dev/seed/sentinel.go`:
- Around line 42-56: The ClickHouse client created via clickhouse.New (variable
ch) is never closed; add a deferred or explicit close for ch when seeding to
ensure the underlying connection is released (the Buffer.Close on
sentinelRequests only drains buffers). Locate the clickhouse.New(...) call that
returns ch and err and add a defer or final ch.Close()/ch.CloseContext(...)
(matching the client's API) after successful creation, ensuring it runs after
sentinelRequests is used and after any errors are handled.
In `@pkg/clickhouse/client.go`:
- Around line 186-195: Client.Close currently just closes c.conn, which drops
pending buffer rows; either enforce buffer closure inside Client or require
callers to close buffers before calling Client.Close. Update the ClickHouse
client implementation (functions NewBuffer and Client.Close) so that NewBuffer
registers created buffers with the Client (e.g., a map/slice of active buffers)
and Client.Close iterates and Close()s each registered buffer (unregistering as
they close), returning any aggregated error; alternatively, make Client.Close
detect active buffers and return an error if any are still open so callers must
Close buffers first. Then update callers creating buffers (notably the
buildSteps and buildStepLogs usages in svc/ctrl/worker/run.go and
svc/api/run.go) to explicitly Close() those buffers before calling Client.Close
to preserve shutdown ordering.
- Around line 52-89: New currently opens a ClickHouse connection with
ch.Open(opts) and if subsequent ping/retry fails it returns an error without
closing conn, leaking the connection; update New so that after ch.Open(opts)
succeeds you ensure conn.Close() is called on all error paths (e.g. if retry.Do
against conn.Ping fails) — either by deferring a conditional close (defer
func(){ if !clientCreated { _ = conn.Close() } }()) or by explicitly calling
conn.Close() before each return that occurs after ch.Open; reference the New
function, the conn variable, ch.Open, retry.Do (conn.Ping) and ensure the
successful return path still returns the open connection without closing it.
---
Duplicate comments:
In `@svc/api/run.go`:
- Around line 162-166: The defer registration order is inverted: because r.Defer
runs LIFO, chClient.Close is currently registered last and will run first
(closing the connection before buffers flush). Fix by registering the client
close before the buffer closes so the client is closed last — i.e., call
r.Defer(chClient.Close) first, then r.Defer for apiRequests.Close,
keyVerifications.Close, and ratelimits.Close (or otherwise ensure chClient.Close
is registered earliest) inside the same function where r.Defer is used.
In `@svc/sentinel/run.go`:
- Around line 141-144: The defer registration order is inverted: because r.Defer
runs handlers LIFO, chClient.Close is currently registered last and thus
executes first, closing the connection before buffer processors flush. Fix by
registering chClient.Close first, then register keyVerifications.Close and
sentinelRequests.Close (so they execute before chClient.Close); reference the
r.Defer calls and the sentinelRequests.Close, keyVerifications.Close, and
chClient.Close symbols when making the change.
---
Nitpick comments:
In `@cmd/dev/seed/local.go`:
- Around line 44-52: The KeyVerifications field in keys.Config is being set to
nil which leaves keys.Config in an invalid state and can panic later; replace
the nil with the package's provided noop batch verifier so the
ClickHouse-disabled path remains safe—update the keys.New call to set
KeyVerifications to the noop batch processor (e.g., the provided
keys.NoopBatchProcessor()/keys.NoopVerifier factory) instead of nil and ensure
any related UsageLimiter/KeyCache/QuotaCache fields follow the same non-nil
pattern if applicable.
In `@cmd/dev/seed/sentinel.go`:
- Around line 273-279: Ensure the sentinel request buffer is closed before the
ClickHouse client is closed so the final flush can complete: explicitly call
s.sentinelRequests.Close() prior to closing the ClickHouse client (ch.Close())
or, if you used defer ch.Close(), remove or delay that defer so it runs after
s.sentinelRequests.Close(); update the code around the log lines and the
functions that manage the lifetime (e.g., s.sentinelRequests.Close() and
ch.Close()) to guarantee the buffer close happens first and add a short comment
documenting this ordering intent.
In `@pkg/batch/process.go`:
- Around line 124-128: Replace the zero-capacity slice allocation with a
capacity hint to avoid repeated growth: change the `batch := make([]T, 0)`
allocation to reserve `bp.config.BatchSize` capacity (e.g., make([]T, 0,
bp.config.BatchSize)) so `batch` starts with sufficient capacity and can still
be reused via `batch[:0]`; update any related comments referencing steady-state
reuse if present.
In `@pkg/buffer/buffer_test.go`:
- Around line 145-146: Replace the testify assert usage with require in this
test: change assert.Equal(t, event, *received, ...) to require.Equal(t, event,
*received) and update the test imports to use
"github.com/stretchr/testify/require" instead of or in addition to assert;
locate the comparison around the channel receive (variable received from b.c) in
buffer_test.go and remove the custom message parameter so it uses require's
signature with t, expected, actual.
In `@svc/api/routes/v2_analytics_get_verifications/200_test.go`:
- Around line 61-62: Replace the brittle time.Sleep(5 * time.Second) with the
same polling pattern used elsewhere in this test file: use
require.EventuallyWithT to repeatedly check for the expected buffered/flush
condition until it becomes true (e.g. a function that queries the verification
buffer/count used by the other checks in this file) with a reasonable timeout
and interval (timeout > 5s, interval ~100–250ms). Locate the sleep call in the
test and swap it for require.EventuallyWithT(t, func(t assert.TestingT) bool {
/* return true when buffer/flush visible */ }, timeout, interval) so the test
aligns with the polling used in the other assertions and avoids flakiness.
🪄 Autofix (Beta)
Fix all unresolved CodeRabbit comments on this PR:
- Push a commit to this branch (recommended)
- Create a new PR with the fixes
ℹ️ Review info
⚙️ Run configuration
Configuration used: Repository UI
Review profile: CHILL
Plan: Pro
Run ID: 5fefc365-a2be-4af6-8380-45d5416fa270
📒 Files selected for processing (63)
cmd/dev/seed/local.gocmd/dev/seed/sentinel.gocmd/dev/seed/verifications.gointernal/services/keys/get.gointernal/services/keys/service.gointernal/services/keys/verifier.gointernal/services/ratelimit/replay.gointernal/services/usagelimiter/redis.gopkg/batch/noop.gopkg/batch/process.gopkg/buffer/buffer.gopkg/buffer/buffer_test.gopkg/clickhouse/billable_ratelimits.gopkg/clickhouse/billable_usage.gopkg/clickhouse/billable_verifications.gopkg/clickhouse/buffer.gopkg/clickhouse/client.gopkg/clickhouse/deployment_request_count.gopkg/clickhouse/flush.gopkg/clickhouse/interface.gopkg/clickhouse/key_last_used.gopkg/clickhouse/noop.gopkg/clickhouse/user.gopkg/zen/middleware_metrics.gosvc/api/internal/testutil/http.gosvc/api/routes/register.gosvc/api/routes/services.gosvc/api/routes/v2_analytics_get_verifications/200_test.gosvc/api/routes/v2_analytics_get_verifications/400_test.gosvc/api/routes/v2_analytics_get_verifications/401_test.gosvc/api/routes/v2_analytics_get_verifications/403_test.gosvc/api/routes/v2_analytics_get_verifications/404_test.gosvc/api/routes/v2_analytics_get_verifications/412_test.gosvc/api/routes/v2_analytics_get_verifications/422_test.gosvc/api/routes/v2_analytics_get_verifications/429_test.gosvc/api/routes/v2_analytics_get_verifications/503_test.gosvc/api/routes/v2_analytics_get_verifications/handler.gosvc/api/routes/v2_keys_verify_key/200_test.gosvc/api/routes/v2_keys_verify_key/400_test.gosvc/api/routes/v2_keys_verify_key/401_test.gosvc/api/routes/v2_keys_verify_key/403_test.gosvc/api/routes/v2_keys_verify_key/404_test.gosvc/api/routes/v2_keys_verify_key/412_test.gosvc/api/routes/v2_keys_verify_key/handler.gosvc/api/routes/v2_keys_verify_key/migration_test.gosvc/api/routes/v2_keys_verify_key/multilimit_test.gosvc/api/routes/v2_keys_verify_key/multiple_ratelimits_overcommit_test.gosvc/api/routes/v2_keys_verify_key/ratelimit_response_test.gosvc/api/routes/v2_keys_verify_key/resend_demo_test.gosvc/api/routes/v2_ratelimit_limit/200_test.gosvc/api/routes/v2_ratelimit_limit/accuracy_test.gosvc/api/routes/v2_ratelimit_limit/handler.gosvc/api/routes/v2_ratelimit_multi_limit/200_test.gosvc/api/routes/v2_ratelimit_multi_limit/handler.gosvc/api/run.gosvc/ctrl/worker/deploy/build.gosvc/ctrl/worker/deploy/service.gosvc/ctrl/worker/run.gosvc/sentinel/config.gosvc/sentinel/middleware/logging.gosvc/sentinel/routes/register.gosvc/sentinel/routes/services.gosvc/sentinel/run.go
💤 Files with no reviewable changes (8)
- svc/api/routes/v2_analytics_get_verifications/503_test.go
- svc/api/routes/v2_analytics_get_verifications/412_test.go
- svc/api/routes/v2_analytics_get_verifications/404_test.go
- svc/api/routes/v2_analytics_get_verifications/403_test.go
- svc/api/routes/v2_analytics_get_verifications/handler.go
- svc/api/routes/v2_analytics_get_verifications/401_test.go
- svc/api/routes/v2_analytics_get_verifications/400_test.go
- pkg/clickhouse/interface.go
✅ Files skipped from review due to trivial changes (3)
- svc/api/routes/v2_keys_verify_key/200_test.go
- svc/api/routes/v2_keys_verify_key/handler.go
- svc/api/routes/v2_keys_verify_key/ratelimit_response_test.go
🚧 Files skipped from review as they are similar to previous changes (32)
- svc/sentinel/routes/register.go
- svc/api/routes/v2_analytics_get_verifications/422_test.go
- internal/services/keys/get.go
- svc/api/routes/v2_keys_verify_key/404_test.go
- svc/api/routes/v2_keys_verify_key/403_test.go
- svc/api/routes/v2_keys_verify_key/multilimit_test.go
- svc/sentinel/routes/services.go
- svc/api/routes/v2_keys_verify_key/401_test.go
- svc/api/routes/v2_keys_verify_key/multiple_ratelimits_overcommit_test.go
- svc/api/routes/v2_keys_verify_key/400_test.go
- svc/api/routes/v2_ratelimit_limit/200_test.go
- svc/api/routes/v2_keys_verify_key/412_test.go
- pkg/clickhouse/user.go
- pkg/clickhouse/key_last_used.go
- svc/api/routes/v2_ratelimit_limit/accuracy_test.go
- pkg/clickhouse/deployment_request_count.go
- pkg/batch/noop.go
- internal/services/keys/verifier.go
- svc/api/routes/services.go
- svc/api/routes/v2_ratelimit_multi_limit/handler.go
- pkg/clickhouse/flush.go
- pkg/clickhouse/billable_verifications.go
- svc/sentinel/middleware/logging.go
- pkg/clickhouse/billable_ratelimits.go
- svc/api/routes/v2_analytics_get_verifications/429_test.go
- pkg/zen/middleware_metrics.go
- cmd/dev/seed/verifications.go
- svc/api/internal/testutil/http.go
- svc/sentinel/config.go
- pkg/clickhouse/buffer.go
- svc/ctrl/worker/deploy/service.go
- internal/services/keys/service.go
Flo4604
force-pushed
the
03-27-chore_rework_clickhouse_package
branch
from
05d1cec to
60ca4f3
Compare
Contributor
There was a problem hiding this comment.
Actionable comments posted: 2
♻️ Duplicate comments (1)
svc/api/internal/testutil/http.go (1)
123-143:⚠️ Potential issue | 🟠 MajorHarness buffers still flush too late.
50_000/5smeans a few test rows stay invisible until the timer fires. That's the root cause of the sleeps/Eventuallyloops in the updated tests. Make harness buffers near-immediate, or add an explicit flush helper.🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@svc/api/internal/testutil/http.go` around lines 123 - 143, The harness buffers (keyVerifications and ratelimitsfer created via clickhouse.NewBuffer with BufferConfig) are configured to batch/flush too slowly causing test visibility delays; change their BufferConfig to near-immediate flush behavior by lowering BatchSize and BufferSize to a small number (e.g., 1 or a low tens) and set FlushInterval to a very short duration (e.g., 10-50ms) and optionally reduce Consumers to 1, or alternately expose/implement and call an explicit Flush method on the buffer in tests after writes; update the BufferConfig for both keyVerifications and ratelimitsfer (and ensure any test cleanup still calls Close) so rows become visible immediately.
🤖 Prompt for all review comments with AI agents
Verify each finding against the current code and only fix it if needed.
Inline comments:
In `@cmd/dev/seed/verifications.go`:
- Around line 60-68: The buffer is created with clickhouse.NewBuffer for
schema.KeyVerification ("keyVerifications") but OnFlushError is nil so async
flush failures are lost; add an OnFlushError callback that records errors (e.g.,
send to a dedicated error channel or append to a shared error slice protected by
mutex) so the seed command can observe them before exiting, and ensure the main
flow waits for Close() completion and checks the collected errors to return a
non-zero result; reference clickhouse.NewBuffer, the keyVerifications variable,
the OnFlushError callback and Close() when implementing the collector and error
propagation.
In `@svc/ctrl/integration/harness/harness.go`:
- Around line 207-208: BuildSteps and BuildStepLogs are set to nil which causes
a panic when the deploy service calls w.buildSteps.Buffer() and
w.buildStepLogs.Buffer(); replace the nils with explicit no-op batch processors
by initializing BuildSteps and BuildStepLogs using batch.NewNoop with the
correct generic types (e.g., schema.BuildStepV1 and schema.BuildStepLogV1) so
calls to Buffer() are safe in tests (use the same fields BuildSteps and
BuildStepLogs and ensure the batch.NewNoop initialization is used instead of
nil).
---
Duplicate comments:
In `@svc/api/internal/testutil/http.go`:
- Around line 123-143: The harness buffers (keyVerifications and ratelimitsfer
created via clickhouse.NewBuffer with BufferConfig) are configured to
batch/flush too slowly causing test visibility delays; change their BufferConfig
to near-immediate flush behavior by lowering BatchSize and BufferSize to a small
number (e.g., 1 or a low tens) and set FlushInterval to a very short duration
(e.g., 10-50ms) and optionally reduce Consumers to 1, or alternately
expose/implement and call an explicit Flush method on the buffer in tests after
writes; update the BufferConfig for both keyVerifications and ratelimitsfer (and
ensure any test cleanup still calls Close) so rows become visible immediately.
🪄 Autofix (Beta)
Fix all unresolved CodeRabbit comments on this PR:
- Push a commit to this branch (recommended)
- Create a new PR with the fixes
ℹ️ Review info
⚙️ Run configuration
Configuration used: Repository UI
Review profile: CHILL
Plan: Pro
Run ID: 86944182-5107-4c7d-a503-fb1f343d54e7
📒 Files selected for processing (83)
cmd/dev/seed/BUILD.bazelcmd/dev/seed/local.gocmd/dev/seed/sentinel.gocmd/dev/seed/verifications.gointernal/services/keys/BUILD.bazelinternal/services/keys/get.gointernal/services/keys/service.gointernal/services/keys/verifier.gointernal/services/ratelimit/replay.gointernal/services/usagelimiter/redis.gopkg/batch/BUILD.bazelpkg/batch/noop.gopkg/batch/process.gopkg/buffer/buffer.gopkg/buffer/buffer_test.gopkg/clickhouse/BUILD.bazelpkg/clickhouse/billable_ratelimits.gopkg/clickhouse/billable_usage.gopkg/clickhouse/billable_verifications.gopkg/clickhouse/buffer.gopkg/clickhouse/client.gopkg/clickhouse/deployment_request_count.gopkg/clickhouse/flush.gopkg/clickhouse/interface.gopkg/clickhouse/key_last_used.gopkg/clickhouse/noop.gopkg/clickhouse/user.gopkg/zen/BUILD.bazelpkg/zen/middleware_metrics.gopkg/zen/middleware_metrics_test.gosvc/api/BUILD.bazelsvc/api/internal/testutil/BUILD.bazelsvc/api/internal/testutil/http.gosvc/api/routes/BUILD.bazelsvc/api/routes/register.gosvc/api/routes/services.gosvc/api/routes/v2_analytics_get_verifications/200_test.gosvc/api/routes/v2_analytics_get_verifications/400_test.gosvc/api/routes/v2_analytics_get_verifications/401_test.gosvc/api/routes/v2_analytics_get_verifications/403_test.gosvc/api/routes/v2_analytics_get_verifications/404_test.gosvc/api/routes/v2_analytics_get_verifications/412_test.gosvc/api/routes/v2_analytics_get_verifications/422_test.gosvc/api/routes/v2_analytics_get_verifications/429_test.gosvc/api/routes/v2_analytics_get_verifications/503_test.gosvc/api/routes/v2_analytics_get_verifications/handler.gosvc/api/routes/v2_keys_verify_key/200_test.gosvc/api/routes/v2_keys_verify_key/400_test.gosvc/api/routes/v2_keys_verify_key/401_test.gosvc/api/routes/v2_keys_verify_key/403_test.gosvc/api/routes/v2_keys_verify_key/404_test.gosvc/api/routes/v2_keys_verify_key/412_test.gosvc/api/routes/v2_keys_verify_key/BUILD.bazelsvc/api/routes/v2_keys_verify_key/handler.gosvc/api/routes/v2_keys_verify_key/migration_test.gosvc/api/routes/v2_keys_verify_key/multilimit_test.gosvc/api/routes/v2_keys_verify_key/multiple_ratelimits_overcommit_test.gosvc/api/routes/v2_keys_verify_key/ratelimit_response_test.gosvc/api/routes/v2_keys_verify_key/resend_demo_test.gosvc/api/routes/v2_ratelimit_limit/200_test.gosvc/api/routes/v2_ratelimit_limit/BUILD.bazelsvc/api/routes/v2_ratelimit_limit/accuracy_test.gosvc/api/routes/v2_ratelimit_limit/handler.gosvc/api/routes/v2_ratelimit_multi_limit/200_test.gosvc/api/routes/v2_ratelimit_multi_limit/BUILD.bazelsvc/api/routes/v2_ratelimit_multi_limit/handler.gosvc/api/run.gosvc/ctrl/integration/harness/harness.gosvc/ctrl/worker/BUILD.bazelsvc/ctrl/worker/deploy/BUILD.bazelsvc/ctrl/worker/deploy/build.gosvc/ctrl/worker/deploy/service.gosvc/ctrl/worker/run.gosvc/krane/internal/sentinel/apply.gosvc/sentinel/BUILD.bazelsvc/sentinel/config.gosvc/sentinel/engine/integration_test.gosvc/sentinel/middleware/BUILD.bazelsvc/sentinel/middleware/logging.gosvc/sentinel/routes/BUILD.bazelsvc/sentinel/routes/register.gosvc/sentinel/routes/services.gosvc/sentinel/run.go
💤 Files with no reviewable changes (9)
- svc/api/routes/v2_analytics_get_verifications/400_test.go
- svc/api/routes/v2_analytics_get_verifications/401_test.go
- svc/api/routes/v2_analytics_get_verifications/403_test.go
- svc/api/routes/v2_analytics_get_verifications/412_test.go
- svc/api/routes/v2_analytics_get_verifications/404_test.go
- svc/api/routes/v2_analytics_get_verifications/handler.go
- svc/api/routes/v2_keys_verify_key/BUILD.bazel
- pkg/clickhouse/interface.go
- svc/api/routes/v2_analytics_get_verifications/503_test.go
✅ Files skipped from review due to trivial changes (17)
- pkg/batch/BUILD.bazel
- svc/api/internal/testutil/BUILD.bazel
- pkg/zen/BUILD.bazel
- svc/api/BUILD.bazel
- cmd/dev/seed/BUILD.bazel
- pkg/batch/noop.go
- svc/api/routes/BUILD.bazel
- svc/sentinel/routes/register.go
- svc/sentinel/routes/BUILD.bazel
- svc/api/routes/v2_keys_verify_key/412_test.go
- svc/api/routes/v2_keys_verify_key/200_test.go
- svc/ctrl/worker/deploy/BUILD.bazel
- svc/sentinel/BUILD.bazel
- pkg/clickhouse/user.go
- svc/sentinel/routes/services.go
- svc/ctrl/worker/BUILD.bazel
- cmd/dev/seed/sentinel.go
🚧 Files skipped from review as they are similar to previous changes (30)
- internal/services/ratelimit/replay.go
- internal/services/usagelimiter/redis.go
- pkg/buffer/buffer_test.go
- pkg/clickhouse/billable_verifications.go
- pkg/clickhouse/flush.go
- pkg/clickhouse/key_last_used.go
- internal/services/keys/get.go
- svc/api/routes/v2_analytics_get_verifications/429_test.go
- svc/api/routes/v2_analytics_get_verifications/422_test.go
- svc/api/routes/v2_keys_verify_key/400_test.go
- svc/api/routes/v2_keys_verify_key/handler.go
- svc/api/routes/v2_keys_verify_key/404_test.go
- svc/api/routes/v2_keys_verify_key/ratelimit_response_test.go
- svc/api/routes/v2_ratelimit_limit/200_test.go
- svc/api/routes/v2_ratelimit_limit/accuracy_test.go
- svc/api/routes/v2_ratelimit_multi_limit/200_test.go
- svc/api/routes/v2_keys_verify_key/multiple_ratelimits_overcommit_test.go
- svc/ctrl/worker/deploy/build.go
- svc/api/routes/v2_keys_verify_key/403_test.go
- svc/api/routes/v2_keys_verify_key/multilimit_test.go
- pkg/clickhouse/billable_ratelimits.go
- pkg/clickhouse/billable_usage.go
- pkg/zen/middleware_metrics.go
- svc/api/routes/services.go
- svc/sentinel/middleware/logging.go
- pkg/clickhouse/buffer.go
- svc/ctrl/worker/deploy/service.go
- svc/sentinel/run.go
- svc/ctrl/worker/run.go
- svc/api/run.go
ogzhanolguncu
approved these changes
Mar 30, 2026
chronark
reviewed
Mar 30, 2026
chronark
reviewed
Mar 30, 2026
chronark
reviewed
Mar 30, 2026
chronark
reviewed
Mar 30, 2026
chronark
reviewed
Mar 30, 2026
chronark
reviewed
Mar 30, 2026
Flo4604
force-pushed
the
03-27-chore_rework_clickhouse_package
branch
from
07d9601 to
c4b7cb1
Compare
Flo4604
force-pushed
the
03-27-chore_rework_clickhouse_package
branch
from
c4b7cb1 to
f1ef56b
Compare
Contributor
There was a problem hiding this comment.
♻️ Duplicate comments (2)
cmd/dev/seed/verifications.go (1)
60-68:⚠️ Potential issue | 🟠 MajorMake this seed buffer strict.
Drop:nilkeeps best-effort mode, andOnFlushError:nilmeans Line 502 still can't fail the command on flush errors. This seed can finish "successfully" after dropped or failed inserts. SetDrop=false, collect flush errors, and return them afterClose().Also applies to: 502-502
🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@cmd/dev/seed/verifications.go` around lines 60 - 68, The seed buffer created by clickhouse.NewBuffer for keyVerifications is currently best-effort because BufferConfig sets Drop:nil and OnFlushError:nil; change BufferConfig to set Drop=false (strict mode) and supply an OnFlushError handler that collects errors into a slice or error aggregator, then after calling keyVerifications.Close() check/return the collected flush errors so the seed command fails on insert failures; update references to BufferConfig.Drop and BufferConfig.OnFlushError and ensure the Close()/defer path returns the aggregated error.svc/sentinel/run.go (1)
145-148:⚠️ Potential issue | 🟠 MajorReverse these defers.
LIFO means
chClient.Close()runs first now. That can break the final buffer flush on shutdown. Registerr.Defer(chClient.Close)before the buffer closers.Minimal fix
- r.Defer(func() error { sentinelRequests.Close(); return nil }) - r.Defer(func() error { keyVerifications.Close(); return nil }) - r.Defer(chClient.Close) + r.Defer(chClient.Close) + r.Defer(func() error { keyVerifications.Close(); return nil }) + r.Defer(func() error { sentinelRequests.Close(); return nil })🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@svc/sentinel/run.go` around lines 145 - 148, The defer registration order is wrong: r.Defer(chClient.Close) must be called before the buffer closers so chClient.Close runs last (after buffers flush). Change the three r.Defer calls so r.Defer(chClient.Close) appears first, then r.Defer(func() error { sentinelRequests.Close(); return nil }) and r.Defer(func() error { keyVerifications.Close(); return nil }) to ensure LIFO executes buffer.Close() before chClient.Close; update the block containing r.Defer, chClient.Close, sentinelRequests.Close, and keyVerifications.Close accordingly.
🧹 Nitpick comments (1)
svc/ctrl/integration/harness/harness.go (1)
209-210: Close no-op processors in test cleanup to avoid goroutine leakage.
NewNoopreturns live processors; in harness tests, close them viat.Cleanup.Minimal patch
+ buildSteps := batch.NewNoop[schema.BuildStepV1]() + buildStepLogs := batch.NewNoop[schema.BuildStepLogV1]() + t.Cleanup(func() { + buildSteps.Close() + buildStepLogs.Close() + }) + deploySvc := deploy.New(deploy.Config{ DB: database, Clickhouse: chClient, @@ - BuildSteps: batch.NewNoop[schema.BuildStepV1](), - BuildStepLogs: batch.NewNoop[schema.BuildStepLogV1](), + BuildSteps: buildSteps, + BuildStepLogs: buildStepLogs, RegistryConfig: deploy.RegistryConfig{URL: "", Username: "", Password: ""},🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@svc/ctrl/integration/harness/harness.go` around lines 209 - 210, NewNoop returns live processors that must be closed in tests to avoid goroutine leaks; update the harness test setup to register t.Cleanup callbacks that close the processors created for BuildSteps and BuildStepLogs (call the processor Close method returned by NewNoop — e.g., call BuildSteps.Close(...) and BuildStepLogs.Close(...) as appropriate, passing a context or handling the error if the Close signature returns one) so both processors are closed during test cleanup.
🤖 Prompt for all review comments with AI agents
Verify each finding against the current code and only fix it if needed.
Duplicate comments:
In `@cmd/dev/seed/verifications.go`:
- Around line 60-68: The seed buffer created by clickhouse.NewBuffer for
keyVerifications is currently best-effort because BufferConfig sets Drop:nil and
OnFlushError:nil; change BufferConfig to set Drop=false (strict mode) and supply
an OnFlushError handler that collects errors into a slice or error aggregator,
then after calling keyVerifications.Close() check/return the collected flush
errors so the seed command fails on insert failures; update references to
BufferConfig.Drop and BufferConfig.OnFlushError and ensure the Close()/defer
path returns the aggregated error.
In `@svc/sentinel/run.go`:
- Around line 145-148: The defer registration order is wrong:
r.Defer(chClient.Close) must be called before the buffer closers so
chClient.Close runs last (after buffers flush). Change the three r.Defer calls
so r.Defer(chClient.Close) appears first, then r.Defer(func() error {
sentinelRequests.Close(); return nil }) and r.Defer(func() error {
keyVerifications.Close(); return nil }) to ensure LIFO executes buffer.Close()
before chClient.Close; update the block containing r.Defer, chClient.Close,
sentinelRequests.Close, and keyVerifications.Close accordingly.
---
Nitpick comments:
In `@svc/ctrl/integration/harness/harness.go`:
- Around line 209-210: NewNoop returns live processors that must be closed in
tests to avoid goroutine leaks; update the harness test setup to register
t.Cleanup callbacks that close the processors created for BuildSteps and
BuildStepLogs (call the processor Close method returned by NewNoop — e.g., call
BuildSteps.Close(...) and BuildStepLogs.Close(...) as appropriate, passing a
context or handling the error if the Close signature returns one) so both
processors are closed during test cleanup.
ℹ️ Review info
⚙️ Run configuration
Configuration used: Repository UI
Review profile: CHILL
Plan: Pro
Run ID: 443998e9-3d30-45f7-8a19-81f40f765829
📒 Files selected for processing (84)
cmd/dev/seed/BUILD.bazelcmd/dev/seed/local.gocmd/dev/seed/sentinel.gocmd/dev/seed/verifications.gointernal/services/keys/BUILD.bazelinternal/services/keys/get.gointernal/services/keys/service.gointernal/services/keys/verifier.gointernal/services/ratelimit/replay.gointernal/services/usagelimiter/redis.gopkg/batch/BUILD.bazelpkg/batch/noop.gopkg/batch/process.gopkg/buffer/buffer.gopkg/buffer/buffer_test.gopkg/clickhouse/BUILD.bazelpkg/clickhouse/billable_ratelimits.gopkg/clickhouse/billable_usage.gopkg/clickhouse/billable_verifications.gopkg/clickhouse/buffer.gopkg/clickhouse/client.gopkg/clickhouse/deployment_request_count.gopkg/clickhouse/flush.gopkg/clickhouse/interface.gopkg/clickhouse/key_last_used.gopkg/clickhouse/noop.gopkg/clickhouse/user.gopkg/zen/middleware_metrics.gopkg/zen/middleware_metrics_test.gosvc/api/BUILD.bazelsvc/api/internal/testutil/BUILD.bazelsvc/api/internal/testutil/http.gosvc/api/routes/BUILD.bazelsvc/api/routes/register.gosvc/api/routes/services.gosvc/api/routes/v2_analytics_get_verifications/200_test.gosvc/api/routes/v2_analytics_get_verifications/400_test.gosvc/api/routes/v2_analytics_get_verifications/401_test.gosvc/api/routes/v2_analytics_get_verifications/403_test.gosvc/api/routes/v2_analytics_get_verifications/404_test.gosvc/api/routes/v2_analytics_get_verifications/412_test.gosvc/api/routes/v2_analytics_get_verifications/422_test.gosvc/api/routes/v2_analytics_get_verifications/429_test.gosvc/api/routes/v2_analytics_get_verifications/503_test.gosvc/api/routes/v2_analytics_get_verifications/handler.gosvc/api/routes/v2_keys_verify_key/200_test.gosvc/api/routes/v2_keys_verify_key/400_test.gosvc/api/routes/v2_keys_verify_key/401_test.gosvc/api/routes/v2_keys_verify_key/403_test.gosvc/api/routes/v2_keys_verify_key/404_test.gosvc/api/routes/v2_keys_verify_key/412_test.gosvc/api/routes/v2_keys_verify_key/BUILD.bazelsvc/api/routes/v2_keys_verify_key/handler.gosvc/api/routes/v2_keys_verify_key/migration_test.gosvc/api/routes/v2_keys_verify_key/multilimit_test.gosvc/api/routes/v2_keys_verify_key/multiple_ratelimits_overcommit_test.gosvc/api/routes/v2_keys_verify_key/ratelimit_response_test.gosvc/api/routes/v2_keys_verify_key/resend_demo_test.gosvc/api/routes/v2_ratelimit_limit/200_test.gosvc/api/routes/v2_ratelimit_limit/BUILD.bazelsvc/api/routes/v2_ratelimit_limit/accuracy_test.gosvc/api/routes/v2_ratelimit_limit/handler.gosvc/api/routes/v2_ratelimit_multi_limit/200_test.gosvc/api/routes/v2_ratelimit_multi_limit/BUILD.bazelsvc/api/routes/v2_ratelimit_multi_limit/handler.gosvc/api/run.gosvc/ctrl/integration/harness/BUILD.bazelsvc/ctrl/integration/harness/harness.gosvc/ctrl/worker/BUILD.bazelsvc/ctrl/worker/deploy/BUILD.bazelsvc/ctrl/worker/deploy/build.gosvc/ctrl/worker/deploy/service.gosvc/ctrl/worker/run.gosvc/krane/internal/sentinel/apply.gosvc/sentinel/BUILD.bazelsvc/sentinel/config.gosvc/sentinel/engine/BUILD.bazelsvc/sentinel/engine/integration_test.gosvc/sentinel/middleware/BUILD.bazelsvc/sentinel/middleware/logging.gosvc/sentinel/routes/BUILD.bazelsvc/sentinel/routes/register.gosvc/sentinel/routes/services.gosvc/sentinel/run.go
💤 Files with no reviewable changes (10)
- svc/api/routes/v2_analytics_get_verifications/412_test.go
- svc/sentinel/engine/BUILD.bazel
- svc/api/routes/v2_analytics_get_verifications/404_test.go
- svc/api/routes/v2_analytics_get_verifications/503_test.go
- svc/api/routes/v2_analytics_get_verifications/403_test.go
- svc/api/routes/v2_keys_verify_key/BUILD.bazel
- svc/api/routes/v2_analytics_get_verifications/401_test.go
- svc/api/routes/v2_analytics_get_verifications/400_test.go
- svc/api/routes/v2_analytics_get_verifications/handler.go
- pkg/clickhouse/interface.go
✅ Files skipped from review due to trivial changes (21)
- svc/api/BUILD.bazel
- pkg/batch/BUILD.bazel
- svc/api/routes/v2_keys_verify_key/401_test.go
- cmd/dev/seed/BUILD.bazel
- svc/ctrl/worker/deploy/BUILD.bazel
- internal/services/keys/BUILD.bazel
- svc/api/internal/testutil/BUILD.bazel
- svc/api/routes/v2_keys_verify_key/412_test.go
- svc/api/routes/BUILD.bazel
- svc/sentinel/routes/register.go
- svc/sentinel/middleware/BUILD.bazel
- svc/api/routes/v2_keys_verify_key/ratelimit_response_test.go
- svc/sentinel/routes/BUILD.bazel
- svc/ctrl/worker/BUILD.bazel
- svc/api/routes/v2_keys_verify_key/404_test.go
- svc/api/routes/v2_ratelimit_limit/accuracy_test.go
- svc/ctrl/integration/harness/BUILD.bazel
- svc/api/routes/v2_keys_verify_key/resend_demo_test.go
- svc/api/routes/v2_keys_verify_key/handler.go
- svc/api/routes/v2_analytics_get_verifications/200_test.go
- internal/services/ratelimit/replay.go
🚧 Files skipped from review as they are similar to previous changes (36)
- cmd/dev/seed/local.go
- internal/services/keys/get.go
- pkg/zen/middleware_metrics_test.go
- pkg/clickhouse/deployment_request_count.go
- svc/api/routes/v2_analytics_get_verifications/422_test.go
- svc/api/routes/v2_keys_verify_key/400_test.go
- svc/api/routes/v2_keys_verify_key/multiple_ratelimits_overcommit_test.go
- pkg/clickhouse/user.go
- svc/sentinel/BUILD.bazel
- pkg/clickhouse/BUILD.bazel
- svc/api/routes/v2_keys_verify_key/migration_test.go
- svc/sentinel/engine/integration_test.go
- svc/ctrl/worker/deploy/build.go
- pkg/clickhouse/key_last_used.go
- svc/api/routes/v2_ratelimit_multi_limit/200_test.go
- svc/api/routes/v2_ratelimit_limit/200_test.go
- pkg/batch/noop.go
- svc/api/routes/v2_analytics_get_verifications/429_test.go
- svc/api/routes/v2_ratelimit_multi_limit/BUILD.bazel
- cmd/dev/seed/sentinel.go
- svc/api/routes/register.go
- pkg/clickhouse/billable_verifications.go
- pkg/clickhouse/billable_ratelimits.go
- pkg/zen/middleware_metrics.go
- svc/sentinel/routes/services.go
- svc/sentinel/middleware/logging.go
- pkg/clickhouse/billable_usage.go
- svc/api/routes/v2_ratelimit_limit/BUILD.bazel
- internal/services/keys/service.go
- svc/ctrl/worker/run.go
- pkg/buffer/buffer.go
- svc/ctrl/worker/deploy/service.go
- internal/services/usagelimiter/redis.go
- pkg/clickhouse/noop.go
- svc/api/run.go
- svc/api/internal/testutil/http.go
chronark
approved these changes
Mar 30, 2026
Contributor
There was a problem hiding this comment.
Actionable comments posted: 1
Caution
Some comments are outside the diff and can’t be posted inline due to platform limitations.
⚠️ Outside diff range comments (1)
cmd/dev/seed/sentinel.go (1)
43-58:⚠️ Potential issue | 🟡 MinorClose the ClickHouse client to prevent resource leak.
The
chclient created at line 43 is passed toNewBufferbut the buffer doesn't take ownership of the connection. The function returns without closingch, leaving the underlying ClickHouse connection open.🐛 Proposed fix
+ defer func() { + if err := ch.Close(); err != nil { + log.Printf("Warning: failed to close ClickHouse connection: %v", err) + } + }() + sentinelRequests := clickhouse.NewBuffer[schema.SentinelRequest](ch, "default.sentinel_requests_raw_v1", clickhouse.BufferConfig{🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@cmd/dev/seed/sentinel.go` around lines 43 - 58, The ClickHouse client 'ch' created by clickhouse.New is not closed, leaking the underlying connection; ensure you call the client's Close method (e.g., defer ch.Close() immediately after successful creation) or otherwise close 'ch' when the function exits/finalizes so the connection is released; update the code around the clickhouse.New(...) error check where 'ch' is assigned and used by clickhouse.NewBuffer (and ensure any Close error is handled/logged).
♻️ Duplicate comments (2)
cmd/dev/seed/verifications.go (1)
60-68:⚠️ Potential issue | 🟠 MajorSeed can exit 0 after losing writes.
Drop: truecan discard rows once the producer outruns ClickHouse, andOnFlushError: nilmeans async insert failures never reach the command. BecauseClose()is void, this path can still print success with fewer rows thannum-verifications. For seeding, make this buffer non-dropping and collect flush errors before returning success.Also applies to: 500-505
svc/api/run.go (1)
168-172:⚠️ Potential issue | 🔴 CriticalFix defer order; client closes first now.
LIFO makes
chClient.Close()run before the bufferClose()calls, so final flushes happen after the connection is gone.pkg/clickhouse/buffer.gorequires closing processors before the client.🐛 Minimal fix
- // Close buffers before connection (LIFO) - r.Defer(func() error { apiRequests.Close(); return nil }) - r.Defer(func() error { keyVerifications.Close(); return nil }) - r.Defer(func() error { ratelimits.Close(); return nil }) - r.Defer(chClient.Close) + // Close connection last (LIFO: first registered closes last) + r.Defer(chClient.Close) + r.Defer(func() error { ratelimits.Close(); return nil }) + r.Defer(func() error { keyVerifications.Close(); return nil }) + r.Defer(func() error { apiRequests.Close(); return nil })🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@svc/api/run.go` around lines 168 - 172, The current defer order causes chClient.Close() to run first (due to LIFO), closing the client before buffers are flushed; to fix, register the client close before registering buffer closes so the buffers' Close() calls run first: call r.Defer(chClient.Close) first, then r.Defer for ratelimits.Close, keyVerifications.Close, and apiRequests.Close (or keep the existing inline wrappers) so the execution order becomes apiRequests.Close -> keyVerifications.Close -> ratelimits.Close -> chClient.Close; reference r.Defer, chClient.Close, apiRequests.Close, keyVerifications.Close, ratelimits.Close and pkg/clickhouse/buffer.go when making the change.
🧹 Nitpick comments (4)
pkg/batch/process.go (1)
142-152: Consider adding nil check for defensive coding (optional).While the current
buffer.Buffer()implementation never sends nil pointers, a defensive nil check would protect against future changes or edge cases.♻️ Proposed defensive check
case ptr, ok := <-c: if !ok { // channel closed t.Stop() if len(batch) > 0 { bp.flush(context.Background(), batch, "close") batch = batch[:0] } return } + if ptr == nil { + continue + } batch = append(batch, *ptr)This matches the pattern used in
internal/services/usagelimiter/redis.go:284-295.🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@pkg/batch/process.go` around lines 142 - 152, Add a defensive nil-pointer check when receiving from the channel in the select branch that handles "case ptr, ok := <-c": if ok and ptr == nil, skip appending (or log/continue) instead of dereferencing *ptr; preserve existing behavior on channel close (t.Stop, flush via bp.flush(context.Background(), batch, "close"), clear batch) and keep appending when ptr is non-nil. This change should be made around the receive logic that currently does "batch = append(batch, *ptr)" to avoid panics if a nil pointer is ever sent.pkg/buffer/buffer.go (1)
105-120: Consider deferring allocation until enqueue succeeds (optional).When
drop=trueand the buffer is full,new(T)allocates memory that's immediately discarded. For high-throughput drop scenarios, this creates unnecessary GC pressure.♻️ Proposed optimization
func (b *Buffer[T]) Buffer(t T) { b.mu.RLock() defer b.mu.RUnlock() if b.isClosed { metrics.BufferState.WithLabelValues(b.name, "closed").Inc() return } - ptr := new(T) - *ptr = t - if b.drop { select { - case b.c <- ptr: + case b.c <- &t: metrics.BufferState.WithLabelValues(b.name, "buffered").Inc() - default: // Emit a metric to signal we dropped a message metrics.BufferState.WithLabelValues(b.name, "dropped").Inc() } } else { + ptr := new(T) + *ptr = t b.c <- ptr metrics.BufferState.WithLabelValues(b.name, "buffered").Inc() } }Note: Using
&tin the drop path is safe since the value escapes to the channel only on success, and the caller'stis copied into the function parameter.🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@pkg/buffer/buffer.go` around lines 105 - 120, The code currently allocates ptr := new(T) and copies *ptr = t before attempting to enqueue, causing wasted allocations when b.drop is true and the channel is full; change the logic to defer allocation by removing ptr := new(T)/*ptr = t at the top and instead allocate or take &t only when the send actually succeeds: in the drop-path use a non-allocating select that tries case b.c <- &t (or allocates a new(T) only on that successful branch) and increments metrics.BufferState.WithLabelValues(b.name, "buffered").Inc() on success and "dropped" on default; in the non-drop path perform the send as before (allocating a pointer for the channel) and increment "buffered". Ensure references to b.drop, b.c and metrics.BufferState.WithLabelValues(b.name, ...) are updated accordingly.svc/sentinel/run.go (1)
126-143: Expose buffer flush failures to ops.Both processors use
OnFlushError: nil, so repeated ClickHouse insert failures are log-only. Wire this into a counter or health signal; otherwise analytics can disappear silently during outages.🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@svc/sentinel/run.go` around lines 126 - 143, The buffers sentinelRequests and keyVerifications are configured with OnFlushError: nil so flush failures are silent; implement a reusable OnFlushError handler and attach it to clickhouse.NewBuffer via the clickhouse.BufferConfig.OnFlushError field for both buffers. The handler should accept the error and context from the buffer, log the error with context, increment a metrics counter (e.g., a new flushFailureCounter labeled by buffer name such as "sentinel_requests" and "key_verifications"), and flip or report a health/alert signal (via your existing health/ops API or a health.setUnhealthy call) so ops can detect persistent failures; update creation of sentinelRequests and keyVerifications to pass this handler into OnFlushError instead of nil.pkg/clickhouse/client.go (1)
139-174: TightenQueryToMapstyping to eliminateany.The method returns
[]map[string]anybut internally stores onlych.Dynamicvalues. This widens the public API unnecessarily. Change the return type to[]map[string]ch.Dynamicor introduce a typed row struct.Only one caller exists (
v2_analytics_get_verifications/handler.go:134), so the refactoring is low-impact. Aligns with the guideline: "Never compromise type safety: Noany".🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@pkg/clickhouse/client.go` around lines 139 - 174, Change QueryToMaps to return []map[string]ch.Dynamic instead of []map[string]any: update the function signature (QueryToMaps) and all local variables/results to use ch.Dynamic for the map values (e.g., results := make([]map[string]ch.Dynamic, 0) and row := make(map[string]ch.Dynamic)), keep scanning into values := make([]ch.Dynamic, len(columns)) and valuePtrs as pointers to those, and propagate the new typed return to its single caller in v2_analytics_get_verifications/handler.go (adjust expected type there). Ensure error wrapping (WrapClickHouseError and fault.Wrap) behavior remains unchanged.
🤖 Prompt for all review comments with AI agents
Verify each finding against the current code and only fix it if needed.
Inline comments:
In `@svc/ctrl/worker/deploy/service.go`:
- Around line 58-59: In New(), make invalid states unrepresentable by guarding
cfg.BuildSteps and cfg.BuildStepLogs against nil: if either cfg.BuildSteps or
cfg.BuildStepLogs is nil, assign a noop batch processor (use
batch.NewNoop[schema.BuildStepV1]() and batch.NewNoop[schema.BuildStepLogV1]())
before assigning to the service fields buildSteps and buildStepLogs so
subsequent calls like buildSteps.Buffer() and buildStepLogs.Buffer() cannot
panic.
---
Outside diff comments:
In `@cmd/dev/seed/sentinel.go`:
- Around line 43-58: The ClickHouse client 'ch' created by clickhouse.New is not
closed, leaking the underlying connection; ensure you call the client's Close
method (e.g., defer ch.Close() immediately after successful creation) or
otherwise close 'ch' when the function exits/finalizes so the connection is
released; update the code around the clickhouse.New(...) error check where 'ch'
is assigned and used by clickhouse.NewBuffer (and ensure any Close error is
handled/logged).
---
Duplicate comments:
In `@svc/api/run.go`:
- Around line 168-172: The current defer order causes chClient.Close() to run
first (due to LIFO), closing the client before buffers are flushed; to fix,
register the client close before registering buffer closes so the buffers'
Close() calls run first: call r.Defer(chClient.Close) first, then r.Defer for
ratelimits.Close, keyVerifications.Close, and apiRequests.Close (or keep the
existing inline wrappers) so the execution order becomes apiRequests.Close ->
keyVerifications.Close -> ratelimits.Close -> chClient.Close; reference r.Defer,
chClient.Close, apiRequests.Close, keyVerifications.Close, ratelimits.Close and
pkg/clickhouse/buffer.go when making the change.
---
Nitpick comments:
In `@pkg/batch/process.go`:
- Around line 142-152: Add a defensive nil-pointer check when receiving from the
channel in the select branch that handles "case ptr, ok := <-c": if ok and ptr
== nil, skip appending (or log/continue) instead of dereferencing *ptr; preserve
existing behavior on channel close (t.Stop, flush via
bp.flush(context.Background(), batch, "close"), clear batch) and keep appending
when ptr is non-nil. This change should be made around the receive logic that
currently does "batch = append(batch, *ptr)" to avoid panics if a nil pointer is
ever sent.
In `@pkg/buffer/buffer.go`:
- Around line 105-120: The code currently allocates ptr := new(T) and copies
*ptr = t before attempting to enqueue, causing wasted allocations when b.drop is
true and the channel is full; change the logic to defer allocation by removing
ptr := new(T)/*ptr = t at the top and instead allocate or take &t only when the
send actually succeeds: in the drop-path use a non-allocating select that tries
case b.c <- &t (or allocates a new(T) only on that successful branch) and
increments metrics.BufferState.WithLabelValues(b.name, "buffered").Inc() on
success and "dropped" on default; in the non-drop path perform the send as
before (allocating a pointer for the channel) and increment "buffered". Ensure
references to b.drop, b.c and metrics.BufferState.WithLabelValues(b.name, ...)
are updated accordingly.
In `@pkg/clickhouse/client.go`:
- Around line 139-174: Change QueryToMaps to return []map[string]ch.Dynamic
instead of []map[string]any: update the function signature (QueryToMaps) and all
local variables/results to use ch.Dynamic for the map values (e.g., results :=
make([]map[string]ch.Dynamic, 0) and row := make(map[string]ch.Dynamic)), keep
scanning into values := make([]ch.Dynamic, len(columns)) and valuePtrs as
pointers to those, and propagate the new typed return to its single caller in
v2_analytics_get_verifications/handler.go (adjust expected type there). Ensure
error wrapping (WrapClickHouseError and fault.Wrap) behavior remains unchanged.
In `@svc/sentinel/run.go`:
- Around line 126-143: The buffers sentinelRequests and keyVerifications are
configured with OnFlushError: nil so flush failures are silent; implement a
reusable OnFlushError handler and attach it to clickhouse.NewBuffer via the
clickhouse.BufferConfig.OnFlushError field for both buffers. The handler should
accept the error and context from the buffer, log the error with context,
increment a metrics counter (e.g., a new flushFailureCounter labeled by buffer
name such as "sentinel_requests" and "key_verifications"), and flip or report a
health/alert signal (via your existing health/ops API or a health.setUnhealthy
call) so ops can detect persistent failures; update creation of sentinelRequests
and keyVerifications to pass this handler into OnFlushError instead of nil.
🪄 Autofix (Beta)
Fix all unresolved CodeRabbit comments on this PR:
- Push a commit to this branch (recommended)
- Create a new PR with the fixes
ℹ️ Review info
⚙️ Run configuration
Configuration used: Repository UI
Review profile: CHILL
Plan: Pro
Run ID: 7e2ffc24-7d93-4a8f-8724-89ad50873765
📒 Files selected for processing (84)
cmd/dev/seed/BUILD.bazelcmd/dev/seed/local.gocmd/dev/seed/sentinel.gocmd/dev/seed/verifications.gointernal/services/keys/BUILD.bazelinternal/services/keys/get.gointernal/services/keys/service.gointernal/services/keys/verifier.gointernal/services/ratelimit/replay.gointernal/services/usagelimiter/redis.gopkg/batch/BUILD.bazelpkg/batch/noop.gopkg/batch/process.gopkg/buffer/buffer.gopkg/buffer/buffer_test.gopkg/clickhouse/BUILD.bazelpkg/clickhouse/billable_ratelimits.gopkg/clickhouse/billable_usage.gopkg/clickhouse/billable_verifications.gopkg/clickhouse/buffer.gopkg/clickhouse/client.gopkg/clickhouse/deployment_request_count.gopkg/clickhouse/flush.gopkg/clickhouse/interface.gopkg/clickhouse/key_last_used.gopkg/clickhouse/noop.gopkg/clickhouse/user.gopkg/zen/middleware_metrics.gopkg/zen/middleware_metrics_test.gosvc/api/BUILD.bazelsvc/api/internal/testutil/BUILD.bazelsvc/api/internal/testutil/http.gosvc/api/routes/BUILD.bazelsvc/api/routes/register.gosvc/api/routes/services.gosvc/api/routes/v2_analytics_get_verifications/200_test.gosvc/api/routes/v2_analytics_get_verifications/400_test.gosvc/api/routes/v2_analytics_get_verifications/401_test.gosvc/api/routes/v2_analytics_get_verifications/403_test.gosvc/api/routes/v2_analytics_get_verifications/404_test.gosvc/api/routes/v2_analytics_get_verifications/412_test.gosvc/api/routes/v2_analytics_get_verifications/422_test.gosvc/api/routes/v2_analytics_get_verifications/429_test.gosvc/api/routes/v2_analytics_get_verifications/503_test.gosvc/api/routes/v2_analytics_get_verifications/handler.gosvc/api/routes/v2_keys_verify_key/200_test.gosvc/api/routes/v2_keys_verify_key/400_test.gosvc/api/routes/v2_keys_verify_key/401_test.gosvc/api/routes/v2_keys_verify_key/403_test.gosvc/api/routes/v2_keys_verify_key/404_test.gosvc/api/routes/v2_keys_verify_key/412_test.gosvc/api/routes/v2_keys_verify_key/BUILD.bazelsvc/api/routes/v2_keys_verify_key/handler.gosvc/api/routes/v2_keys_verify_key/migration_test.gosvc/api/routes/v2_keys_verify_key/multilimit_test.gosvc/api/routes/v2_keys_verify_key/multiple_ratelimits_overcommit_test.gosvc/api/routes/v2_keys_verify_key/ratelimit_response_test.gosvc/api/routes/v2_keys_verify_key/resend_demo_test.gosvc/api/routes/v2_ratelimit_limit/200_test.gosvc/api/routes/v2_ratelimit_limit/BUILD.bazelsvc/api/routes/v2_ratelimit_limit/accuracy_test.gosvc/api/routes/v2_ratelimit_limit/handler.gosvc/api/routes/v2_ratelimit_multi_limit/200_test.gosvc/api/routes/v2_ratelimit_multi_limit/BUILD.bazelsvc/api/routes/v2_ratelimit_multi_limit/handler.gosvc/api/run.gosvc/ctrl/integration/harness/BUILD.bazelsvc/ctrl/integration/harness/harness.gosvc/ctrl/worker/BUILD.bazelsvc/ctrl/worker/deploy/BUILD.bazelsvc/ctrl/worker/deploy/build.gosvc/ctrl/worker/deploy/service.gosvc/ctrl/worker/run.gosvc/krane/internal/sentinel/apply.gosvc/sentinel/BUILD.bazelsvc/sentinel/config.gosvc/sentinel/engine/BUILD.bazelsvc/sentinel/engine/integration_test.gosvc/sentinel/middleware/BUILD.bazelsvc/sentinel/middleware/logging.gosvc/sentinel/routes/BUILD.bazelsvc/sentinel/routes/register.gosvc/sentinel/routes/services.gosvc/sentinel/run.go
💤 Files with no reviewable changes (10)
- svc/sentinel/engine/BUILD.bazel
- svc/api/routes/v2_analytics_get_verifications/404_test.go
- svc/api/routes/v2_analytics_get_verifications/403_test.go
- svc/api/routes/v2_analytics_get_verifications/503_test.go
- svc/api/routes/v2_analytics_get_verifications/412_test.go
- svc/api/routes/v2_analytics_get_verifications/handler.go
- svc/api/routes/v2_keys_verify_key/BUILD.bazel
- pkg/clickhouse/interface.go
- svc/api/routes/v2_analytics_get_verifications/401_test.go
- svc/api/routes/v2_analytics_get_verifications/400_test.go
✅ Files skipped from review due to trivial changes (21)
- svc/ctrl/worker/deploy/BUILD.bazel
- internal/services/keys/BUILD.bazel
- svc/sentinel/middleware/BUILD.bazel
- svc/api/routes/BUILD.bazel
- pkg/batch/BUILD.bazel
- svc/sentinel/BUILD.bazel
- svc/ctrl/integration/harness/BUILD.bazel
- svc/sentinel/routes/BUILD.bazel
- svc/api/routes/v2_keys_verify_key/404_test.go
- svc/api/routes/v2_keys_verify_key/400_test.go
- svc/ctrl/worker/BUILD.bazel
- svc/api/BUILD.bazel
- svc/api/routes/v2_keys_verify_key/412_test.go
- svc/api/routes/v2_ratelimit_limit/BUILD.bazel
- pkg/zen/middleware_metrics_test.go
- svc/api/internal/testutil/BUILD.bazel
- svc/sentinel/config.go
- pkg/clickhouse/flush.go
- svc/api/routes/v2_analytics_get_verifications/200_test.go
- cmd/dev/seed/BUILD.bazel
- internal/services/keys/get.go
🚧 Files skipped from review as they are similar to previous changes (33)
- internal/services/ratelimit/replay.go
- svc/api/routes/v2_keys_verify_key/401_test.go
- svc/sentinel/routes/register.go
- svc/api/routes/v2_keys_verify_key/200_test.go
- cmd/dev/seed/local.go
- svc/api/routes/v2_analytics_get_verifications/422_test.go
- pkg/batch/noop.go
- pkg/clickhouse/BUILD.bazel
- svc/krane/internal/sentinel/apply.go
- svc/api/routes/v2_keys_verify_key/migration_test.go
- svc/sentinel/engine/integration_test.go
- svc/api/routes/v2_keys_verify_key/ratelimit_response_test.go
- svc/api/routes/v2_keys_verify_key/multilimit_test.go
- pkg/clickhouse/billable_ratelimits.go
- pkg/clickhouse/deployment_request_count.go
- svc/api/routes/v2_ratelimit_multi_limit/200_test.go
- pkg/clickhouse/user.go
- svc/sentinel/middleware/logging.go
- svc/api/routes/v2_keys_verify_key/handler.go
- pkg/clickhouse/key_last_used.go
- svc/api/routes/v2_analytics_get_verifications/429_test.go
- svc/api/routes/v2_ratelimit_multi_limit/BUILD.bazel
- svc/api/internal/testutil/http.go
- pkg/zen/middleware_metrics.go
- pkg/clickhouse/billable_usage.go
- svc/api/routes/v2_ratelimit_limit/handler.go
- svc/ctrl/worker/run.go
- pkg/clickhouse/billable_verifications.go
- svc/api/routes/v2_ratelimit_limit/accuracy_test.go
- internal/services/usagelimiter/redis.go
- pkg/clickhouse/buffer.go
- pkg/clickhouse/noop.go
- svc/ctrl/worker/deploy/build.go
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
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
3 participants
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.
What does this PR do?
Refactors ClickHouse batch processing from a monolithic client with built-in buffers to a modular system using external batch processors. This change separates concerns by moving batch processing logic out of the ClickHouse client and into dedicated, type-safe batch processors that can be created on-demand.
The refactoring introduces:
batch.NewNoop[T]()function for creating no-op batch processors when ClickHouse is not configuredclickhouse.NewBuffer[T]()function that creates type-safe batch processors wired to a ClickHouse clientKeyVerificationsbatch processor instead of a raw ClickHouse clientType of change
How should this be tested?
Checklist
Required
pnpm buildpnpm fmtmake fmton/godirectoryconsole.logsgit pull origin mainAppreciated