Skip to content

Add fire-and-forget async statement submission - #198

Merged
Ch4s3 merged 3 commits into
masterfrom
feat/async-statement-submission
Sep 24, 2026
Merged

Ch4s3 merged 3 commits into
masterfrom
feat/async-statement-submission

Conversation

@Ch4s3

@Ch4s3 Ch4s3 commented Sep 23, 2026 •

Copy link
Copy Markdown
Contributor

Summary

Adds fire-and-forget async statement submission so a caller can start a Snowflake statement, get its handle back without holding a connection pool slot for the statement's full duration, and retrieve its rows later.

Today Snowflex.Transport.Http polls Snowflake's 202 response to completion inside the GenServer.call that owns the checked-out connection (await_async_execution/3), so a slot stays occupied until the statement finishes. Snowflake's SQL API already supports async=true, which acknowledges with a statementHandle as soon as it accepts the statement — Snowflex just never set it.

{:ok, handle} = Snowflex.submit_async(MyRepo, "CALL rebuild_scenarios(?)", [scenario_id])

# fetch_result/3 answers {:ok, :running} until it finishes, so it can poll alone
case Snowflex.fetch_result(MyRepo, handle) do
  {:ok, :running} -> :still_running
  {:ok, %Snowflex.Result{rows: rows}} -> rows
  {:error, error} -> Logger.error(Exception.message(error))
end

# or, if you don't care about rows
{:ok, :succeeded} = Snowflex.statement_status(MyRepo, handle)
:ok = Snowflex.cancel_statement(MyRepo, handle)

Rows come back decoded exactly as Repo.query/3 decodes them, because fetch_result/3 reuses the existing gather_results/3 + format_response_body/1 machinery (including parallel multi-partition merging).

Design notes

Routed through the pool, not around it. Since async=true returns in milliseconds, going through the pool costs only the submission round-trip while keeping JWT refresh, retry/backoff, Finch pool routing, telemetry and log metadata.

Reuses the execute path. Snowflex.Query gains an :op field; Connection.handle_execute/4 dispatches on it. For status/fetch/cancel, :statement carries the handle so it lands in snowflex_statement log metadata — deliberate, so a stuck statement is greppable. Happy to split it into a separate field if you'd prefer.

Transport callbacks are optional, so existing custom transports still satisfy the behaviour with no change and no compile warnings; calling one that's missing returns a descriptive Snowflex.Error.

:current_statement is untouched by the async path — that field belongs to the cursor/streaming path, and writing it would corrupt an in-flight Repo.stream on the same connection.

Included bug fixes

Two pre-existing unit mismatches, both in code this PR builds on. I'd originally planned to split these out, but the first one directly undermines the feature — for a fire-and-forget statement the server-side timeout is the only bound, since nothing client-side waits on it.

  • The statement request body sent timeout in milliseconds where Snowflake defines seconds, so the 45s default asked Snowflake to wait ~12.5 hours, effectively disabling the server-side statement timeout. Behaviour change: statements that previously ran unbounded will now be cancelled at the configured timeout. Called out in the CHANGELOG.
  • JWT exp added a millisecond :token_lifetime to a second-precision timestamp, advertising ~7 days instead of the configured 10 minutes. Verified fixed: exp - iat = 600s.

Also: the JWT is now cached for its lifetime instead of being re-signed (PEM decode + RSA) on every request — verified 1 distinct token across 4 requests, previously 4.

Review round

I ran an adversarial review over the first two commits (9 finder angles, 5 verifiers, mutation testing). 03f7fcf fixes what it found. The ones worth calling out:

  • Async ops inside Repo.checkout/2 silently corrupted the enclosing checkout. run_op/5 asked the pool for a second connection instead of reusing the held one; at pool_size: 1 that queued until the timeout fired and the outer checkout was force-disconnected — while still returning {:ok, handle}, so it looked fine. Reproduced at 5.0s + disconnect, now 0.07s clean. Fixed by reusing the checked-out connection the way stream_query/5 already did.
  • Timeouts raised instead of returning errors. The new entry points lacked the try/catch execute_statement/4 has, so an uncaught exit escaped handle_execute and DBConnection re-raised it. Note declare/fetch also lack it, so execute_statement was the outlier — worth a follow-up to make all six consistent.
  • Status polls could block ~187s against a 45s deadline, because receive_timeout equalled the call timeout and Req retried the GET. Retries are now off for status (Snowflake already answered with 202) and the call deadline has grace over the HTTP timeout.
  • No way to read the rows — the original motivation for the whole change. Hence fetch_result/3.
  • The pool-occupancy test passed for the wrong reason. Mutation testing proved that reintroducing the polling bug left its central assertion green, since submit_async is awaited before the second query starts. It now asserts on submit's own elapsed time, which does fail under that mutation.

Two candidate findings were refuted by testing and deliberately not "fixed": Code.ensure_loaded? is load-bearing (proved by purging the module) and its embedded-mode failure path is unreachable for a running module; and params: %{} vs no params is byte-identical on the wire (Finch normalizes "" and nil) — tidied to [] anyway.

Test plan

  • mix test --exclude integration — 159 passed (25 async tests)
  • mix compile --warnings-as-errors
  • mix credo --strict — no issues
  • mix format --check-formatted
  • mix assay (dialyzer) — 0 errors
  • Mutation-tested the pool guard: reintroducing polling fails the test on both the elapsed-time assertion and refute_received
  • Empirically verified the checkout fix, the timeout path, JWT caching and exp
  • Integration tests against a live Snowflake account — not run; the 68 failures in HttpRepoTest/ConnectionTest are credential-dependent and fail identically on master (verified by diffing failing test names — the sets match)

The statement-timeout fix in particular deserves a live check before release, since it changes how long Snowflake lets a statement run.

Note on the linked issue

Two API names in #197 don't exist: there is no Snowflex.query_async/2 (the real mechanism is the implicit 202 polling above), and no Snowflex.HttpClient.transport_options/1 — the function consumers actually use is Snowflex.Transport.Http.options/1, which is public and documented. So the issue's "coupled to Snowflex internals" concern is unfounded. I'll follow up on the issue to correct both.

Refs #197

Submitting a statement through Snowflex holds a connection pool slot for the
statement's entire run: the transport polls Snowflake's 202 response to
completion inside the GenServer.call that owns the checked-out connection. For
work where the caller only needs to start a statement and never reads its rows
(stored procedures, long MERGEs, warehouse maintenance), that slot is occupied
for no benefit and blocks other queries from acquiring it.

Snowflake's SQL API already supports this directly via the `async=true` query
parameter, which acknowledges with a statement handle as soon as it accepts the
statement. Snowflex never set it. Thread it through and expose the lifecycle as
public API, so consumers stop reimplementing this against Transport.Http.options/1.

Submission is routed through the pool rather than around it, so these calls keep
JWT refresh, retry/backoff, Finch pool routing, telemetry and log metadata. The
slot is now held only for the submission round-trip.

The three transport callbacks are optional so existing custom transports still
satisfy the behaviour; calling them on a transport that lacks them returns a
Snowflex.Error instead of raising UndefinedFunctionError.

Refs #197
Code.ensure_loaded?/1 returns a boolean that was being discarded, which read as
though the call were there for its side effect alone. Using it as a guard also
short-circuits function_exported?/3 for a transport that cannot be loaded.
Adversarial review of the previous two commits turned up a set of defects,
several of which were only provable empirically. Fixes, roughly by severity:

Async ops inside Ecto.Repo.checkout/2 asked the pool for a second connection
instead of reusing the one the caller already held. At pool_size: 1 that queued
behind the caller's own connection until the queue timeout fired and the outer
checkout was force-disconnected — while still returning {:ok, handle}, so the
corruption was silent. run_op/5 now reuses a checked-out connection the way
stream_query/5 already did (5.0s + disconnect -> 0.07s clean).

The three new transport entry points called GenServer.call without the try/catch
execute_statement/4 has, so a call timeout let an uncaught exit escape
handle_execute; DBConnection re-raised it and the caller never saw a
Snowflex.Error. They now catch, and the call deadline gets a grace period over
the HTTP timeout so a slow request surfaces as the handler's own error instead of
racing the deadline. Status polls also no longer inherit Req's transient retry,
which could otherwise block ~187s against a 45s deadline re-asking a question
Snowflake had already answered with 202.

There was no way to read an async statement's rows: the 200 response carries
data and resultSetMetaData, but those were discarded. Consumers would have had to
hand-roll HTTP against Transport.Http.options/1 — the duplication this work set
out to remove. Adds fetch_result/3, reusing gather_results/3 and
format_response_body/1 so rows decode and multi-partition results merge exactly
as they do for a normal query.

run_op/5 built %Query{} directly, skipping the IO.iodata_to_binary normalization
Query.new/1 applies, so an iodata statement was JSON-encoded as an array rather
than a string. A null statementHandle also satisfied the success pattern and
produced {:ok, nil}, which then failed statement_status/3's is_binary guard.

The statement request body sent :timeout in milliseconds where Snowflake defines
seconds, so the 45s default asked for ~12.5 hours and effectively disabled the
server-side timeout — the only bound on a fire-and-forget statement. JWT exp had
the same unit mismatch, advertising ~7 days instead of the configured lifetime.
Both now convert; the token is also cached for its lifetime rather than being
re-signed (PEM decode + RSA) on every request.

Smaller: statement_status/3 reports status positively instead of inferring
:succeeded by exhaustion, so it no longer depends on Result.metadata being a bare
map; error building is unified and keeps the :statement metadata that aids
debugging; Error.code is always a string; handle-addressed GETs share one URL
builder; :hibernate is consistent across reply paths; and Req gets [] rather than
%{} for "no query params", which is the value it treats as a no-op.

The pool-occupancy test was passing for the wrong reason: mutation testing showed
that reintroducing the polling bug left its central assertion green, because
submit_async is awaited before the second query starts. It now asserts on
submit's own elapsed time, which does fail under that mutation. Also covers the
timeout path, checkout reuse, fetch_result, iodata, and the null handle.

@jeremyowensboggs jeremyowensboggs left a comment

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I think this looks great!

# encode path instead.
query_params = if call_opts[:async?], do: %{async: true}, else: []

case Req.post(req_client,

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

We should consider adding retry: false here as well - to prevent a transient error from causing a re-submission. https://github.com/pepsico-ecommerce/galileo/blob/master/apps/api_services/lib/api_services/common/snowflake/async_statement.ex#L244

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Should we use Snowflake's requestId for that? Have callers supply one if they want idempotent requests?

Iirc Req does not automatically retry POST

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

hmm -yup you are correct - it's not on by default http://req.hexdocs.pm/0.7.2/Req.Steps.html#retry/1-request-options.

Withdrawn.

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

So we're good?

@Ch4s3
Ch4s3 marked this pull request as ready for review September 24, 2026 13:29
@Ch4s3
Ch4s3 merged commit caa46e1 into master Sep 24, 2026
7 checks passed
@Ch4s3 Ch4s3 mentioned this pull request Sep 24, 2026
3 of 4 tasks
imorales-g2 pushed a commit to imorales-g2/snowflex that referenced this pull request Sep 30, 2026
Cuts the Unreleased section to 1.7.0. The async statement submission work
(pepsico-ecommerce#198) and the Req 0.7 minimum bump (pepsico-ecommerce#194) both landed since 1.6.0.

Promotes the statement-timeout unit fix from Bug Fixes to Breaking Changes:
sending :timeout in milliseconds where Snowflake expects seconds had
effectively disabled the server-side timeout, so statements that previously
ran unbounded will now be cancelled. That is a behaviour change consumers
need to see before upgrading, not a quiet fix.
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment

Labels

None yet

Projects

None yet

Development

Successfully merging this pull request may close these issues.

3 participants