queuefs: add SQL backend coverage for SQLite, TiDB, and PostgreSQL - #14
Open
JaySon-Huang wants to merge 13 commits into
Open
queuefs: add SQL backend coverage for SQLite, TiDB, and PostgreSQL#14JaySon-Huang wants to merge 13 commits into
JaySon-Huang wants to merge 13 commits into
Conversation
andypeng2015
pushed a commit
to andypeng2015/agfs
that referenced
this pull request
Jun 5, 2026
Both the Python and Go SDKs disabled HTTP timeouts entirely for
streaming endpoints (`timeout=None` and `Timeout: 0` respectively) so
that long-but-well-behaved streams wouldn't be killed by a fixed
overall deadline. The unintended consequence was that a server that
stopped sending bytes mid-stream could hang the client indefinitely —
including any agent or FUSE mount that depends on these SDKs.
This change introduces a configurable per-chunk *inactivity* timeout
that bounds inter-byte silence without bounding total stream duration.
Python SDK (`agfs-sdk/python/pyagfs/client.py`)
- `AGFSClient.__init__` gains `streaming_progress_timeout`, default
60s. The class exposes `DEFAULT_STREAMING_PROGRESS_TIMEOUT` for
callers that want to refer to it explicitly.
- New private helper `_streaming_timeout()` returns the
`(connect_timeout, read_timeout)` tuple `requests` expects, or
`None` if the user explicitly opted out (matches pre-2026-05
behaviour).
- All three streaming call sites switched off `timeout=None`:
`cat(stream=True)`, `write(streaming data)`, and `grep(stream=True)`.
The `read` leg of `requests`' timeout tuple is the per-chunk
inactivity timeout, which is exactly the bound we want — `requests`
handles the implementation natively.
Go SDK (`agfs-sdk/go/client.go`)
- `Client` gains `streamingProgressTimeout`, initialized from the new
`DefaultStreamingProgressTimeout` (60s) in both constructors.
`SetStreamingProgressTimeout(d)` lets callers override or opt out.
- New `progressReader` wraps streaming response bodies. Each
successful Read signals progress to a watchdog goroutine on a
buffered channel; the watchdog resets a `time.Timer` on each
signal. If no progress signal arrives within the configured
timeout, the watchdog calls the request context's `CancelFunc`,
closing the connection and surfacing a read error to the caller.
Setting the timeout to `<=0` skips watchdog setup entirely — the
reader is then a pure passthrough that still cancels the request
context on Close so HTTP resources are released cleanly.
- `ReadStream` and `ReadHandleStream` now build their request with
`http.NewRequestWithContext`, returning `progressReader` instead of
the raw `resp.Body`. The streaming `http.Client.Timeout` stays at
`0` (no overall deadline) — that responsibility moves to the
per-chunk watchdog.
Tests
Python — `agfs-sdk/python/tests/test_streaming_progress_timeout.py`
(new):
- `test_streaming_read_progress_timeout_fires`: a local HTTP server
flushes the headers and then sleeps. With
`streaming_progress_timeout=0.3`, iterating the body raises a
`RequestException` carrying "timed out" within sub-second time.
The test pins the user-visible contract (any timeout-flavored
RequestException) rather than the exact subclass, because
`requests` wraps urllib3's `ReadTimeoutError` differently depending
on transport state (chunked vs. content-length).
- `test_streaming_timeout_opt_out_preserves_legacy_behaviour`:
`streaming_progress_timeout=None` matches the old
`timeout=None` and the consume thread stays blocked until the
fixture stops the server.
- `test_streaming_timeout_tuple_uses_connect_and_progress_legs`:
pins `_streaming_timeout()` shape so a future refactor can't
silently merge the two timeouts back into one.
Go — `agfs-sdk/go/streaming_progress_timeout_test.go` (new):
- `TestStreamingProgressTimeout_FiresOnStalledServer`: handler binds
on `r.Context().Done()` so the test server shuts down cleanly when
the progressReader cancels. With a 300ms progress timeout,
ReadStream errors in <500ms instead of waiting the 3s stall.
- `TestStreamingProgressTimeout_PassesThroughWhenDisabled`:
`SetStreamingProgressTimeout(0)` lets a 250ms gap before any bytes
succeed.
- `TestStreamingProgressTimeout_ProgressKeepsAlive`: 3 chunks 200ms
apart succeed under a 500ms per-chunk bound — the cumulative >1s
stream is fine because each *gap* fits inside the bound.
- `TestStreamingProgressTimeout_DefaultIsSet`: pins both
constructors to `DefaultStreamingProgressTimeout`.
Verification:
# Python
cd agfs-sdk/python
PYTHONPATH=. .venv/bin/python -m pytest tests/ --timeout=15
# → 3 passed in ~10s
# Go
cd agfs-sdk/go
go test ./... -timeout 30s # → all PASS in <1.2s
go test -race -run TestStreamingProgressTimeout -timeout 30s
# → PASS, race clean
Closes the SDK half of the diagnostics P0 ("add SDK streaming progress
timeouts"). The FUSE half (task c4pt0r#14) is the bounded cache PR on
`fix/agfs-fuse-bounded-cache`.
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
1 participant
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.
Summary
Testing