stall

package
v0.56.0 Latest Latest
Warning

This package is not in the latest version of its module.

Go to latest
Published: Oct 11, 2026 License: MIT Imports: 16 Imported by: 0

Documentation

Overview

Package stall bounds how long a storage driver waits for a backend that does not answer, without ever cutting a transfer that keeps moving.

It started as the S3 driver's own answer to issue #44 (a drop into a dead bucket took 85.6 s to say so, a store that accepted the connection and never answered did not return at all) and moved here for issue #73, when the same gaps turned up in the other network drivers: WebDAV cut every transfer at 60 s, whether it was moving or not, and FTP, SFTP and SMB waited forever on a server that stopped answering once connected. One set of pieces now serves them all:

  • Policy: a storage's three settings (attempt timeout, attempts, total budget), read from its config, and the descriptor fields that draw them on the admin form (Defaults.Fields).
  • Conn, Reader, AnswerWatch: the silence limits. Each renews on every sign of life, so it bounds a stall and never the length of a transfer.
  • Clock and Policy.Do: the retry budget.
  • Policy.Reason and UnavailableError: "the store is down" in words, as storage.ErrUnavailable.

⚠ The limits are on SILENCE. A 2 GB upload on a slow line takes as long as it takes; what is bounded is how long any one wait for a sign of life may last. A test that only proves "the dead store is reported in time" proves half of it — every driver here also has a test that a transfer many times the attempt timeout long, against a store that reads or writes slowly, arrives whole.

Index

Constants

View Source
const (
	MaxAttemptsLimit     = 20
	AttemptTimeoutLimitS = 600
	TotalTimeoutLimitS   = 3600
)

Bounds of the three settings, the same for every driver. The descriptor's Min/Max read these constants, so the form and the driver cannot disagree.

View Source
const MaxBackoff = 10 * time.Second

MaxBackoff caps the pause between two attempts.

Variables

View Source
var AttemptText = FieldText{
	Label:       "Attempt timeout (seconds)",
	I18nKey:     "storages.fields.attemptTimeout",
	Help:        "How long one attempt may wait for a sign of life from the server: connecting, logging in, the answer to each request, and each next piece of a transfer. A transfer that keeps moving is never cut, however long it takes; an upload the server stops taking is cut after 60 seconds, or after this if it is longer.",
	HelpI18nKey: "storages.fieldHelp.attemptTimeout",
}

AttemptText is the attempt-timeout text of ServerTexts for a driver whose server answers every request at once (FTP, SFTP, SMB).

View Source
var SendStallFloor = 60 * time.Second

SendStallFloor is the least time a send may go without moving before it is cut, whatever the attempt timeout.

⚠ A send does not move smoothly. The kernel wakes a writer blocked on a full send buffer only when about half of what it holds has drained, so against a store that reads slowly a healthy upload sits still for (queued/2)/rate at a time — measured 1-2 s at 1 MB/s with Linux's 4 MB send buffer, which a 1 s attempt timeout cut in half the runs. A variable so a test can lower it.

Functions

func CloseRaw

func CloseRaw(c net.Conn)

CloseRaw closes a connection under TLS without a close_notify, which would wait on a peer that is not answering.

func Closed

func Closed(err error) error

Closed marks err (an io.EOF on a command connection) as the server having closed the connection. Not every EOF means that — a body that ends early is read as EOF too, which is why Reason does not take a bare EOF for it — so the driver that knows says so.

func EndedInSilence

func EndedInSilence(err error) bool

EndedInSilence: the attempt was cut by a timeout — ours (Error) or the network's (a connect, a lookup, a send that timed out) — not broken while it moved.

func IntSetting

func IntSetting(v any, def, hi int) int

IntSetting reads one positive integer setting.

func RefusedByTLS

func RefusedByTLS(err error) bool

RefusedByTLS: a certificate that does not verify, or a TLS endpoint that does not speak TLS, is a misconfiguration, not a network hiccup. Asking again cannot change the answer, so it is never retried.

func SendTail

func SendTail(n int64) time.Duration

SendTail is the extra answer wait for a request body of n bytes.

Types

type Activity added in v0.50.0

type Activity struct {
	// contains filtered or unexported fields
}

Activity bounds silence on a connection a library reads from in the BACKGROUND - SSH's multiplexer under the SFTP driver, the receiver goroutine under the SMB driver (issue #75).

⚠ Conn cannot do it. Conn sets a deadline on every Read, and those libraries keep one Read waiting for as long as the session is open: on a session that sits idle between two operations, that Read is waiting for nothing, and a deadline on it kills a healthy session. So the limit is armed per CALL instead (Watch): while an operation runs, the peer must show a sign of life - a byte arrives, or, for a send, a byte leaves - at least once every limit; when it does not, the connection is cut (Cut), which fails the operation and everything else waiting on that connection, and the driver opens a new one.

The clock runs only while a watched call is under way, so an idle session is never cut, and it restarts on every byte, so a transfer that keeps moving is never cut either.

func Track added in v0.50.0

func Track(c net.Conn) (*Activity, net.Conn)

Track wraps c: hand the returned connection to the library and keep the Activity to watch its calls with.

func (*Activity) Cut added in v0.50.0

func (a *Activity) Cut()

Cut closes the connection: every call waiting on it fails at once.

func (*Activity) WasCut added in v0.50.0

func (a *Activity) WasCut() bool

WasCut reports whether the connection was cut.

func (*Activity) Watch added in v0.50.0

func (a *Activity) Watch(limit time.Duration, sends bool) *CallWatch

Watch bounds the silence of one call: if, while it runs, nothing arrives from the peer for limit - nor, when sends, does the peer take anything we send - the connection is cut. Stop the watch when the call returns.

type AnswerWatch

type AnswerWatch struct {
	// contains filtered or unexported fields
}

AnswerWatch closes the connection when the answer to a sent HTTP request does not start within limit. Closing, not cancelling: a cancelled context would also end the body of a download the caller has not read yet.

Install Trace on the request's context, and call Stop when the transport hands over the answer — for a plain http.RoundTripper, when RoundTrip returns.

⚠ NOT at httptrace's GotFirstResponseByte. That hook fires once, on the first byte of ANY response, and an upload sent with "Expect: 100-continue" (the AWS SDK does it over 2 MB) gets an interim "100 Continue" before the body is even sent: a watch stopped there never runs, and a store that took the whole upload and then never answered was waited for forever.

func NewAnswerWatch

func NewAnswerWatch(limit time.Duration) *AnswerWatch

NewAnswerWatch watches one attempt.

func (*AnswerWatch) Fired

func (w *AnswerWatch) Fired() bool

Fired reports whether the watch cut the attempt.

func (*AnswerWatch) Limit

func (w *AnswerWatch) Limit() time.Duration

Limit is the wait the watch allows.

func (*AnswerWatch) Stop

func (w *AnswerWatch) Stop()

Stop ends the watch: the answer has arrived (or the attempt is over).

func (*AnswerWatch) Trace

func (w *AnswerWatch) Trace() *httptrace.ClientTrace

Trace is the hooks that follow the request.

type CallWatch added in v0.50.0

type CallWatch struct {
	// contains filtered or unexported fields
}

CallWatch is one call's silence limit (Activity.Watch).

func (*CallWatch) Limit added in v0.50.0

func (w *CallWatch) Limit() time.Duration

Limit is the silence the watch allows.

func (*CallWatch) Stop added in v0.50.0

func (w *CallWatch) Stop() bool

Stop ends the watch and reports whether it cut the connection.

type Clock

type Clock struct {
	P        Policy
	Began    time.Time // when the operation started (for the message)
	Attempts int       // attempts started so far
	// contains filtered or unexported fields
}

Clock is one operation's time account, shared by its attempts (which run one after another, never at once).

func NewClock

func NewClock(p Policy) *Clock

NewClock starts an operation's clock.

func (*Clock) Failed

func (c *Clock) Failed(started time.Time, err error)

Failed records a failed attempt that started at started. An attempt that ran longer than the whole budget was a transfer that moved and then broke — a store that is down gives no sign of life for one attempt timeout and is cut there — so the budget starts over from its failure, and a 2 GB upload that loses its connection near the end is still retried.

⚠ Not when the attempt ended in silence that a guard cut: a send the store stopped taking (SendStall, a minute) or an answer that never came. That attempt ran long because the store was silent, and a fresh budget would buy the same minute again, MaxAttempts times over.

func (*Clock) Room

func (c *Clock) Room() time.Duration

Room is how long the retryer may still wait before the next attempt so that a store that is still down could show it inside the budget. That costs what the last attempt did, up to one attempt timeout: a refused port fails in a millisecond and may be retried until the budget is spent, a silent store costs a whole attempt timeout every time. Negative: no more.

⚠ Capped at the attempt timeout, not at the longest the last attempt could have waited. An upload may wait the attempt timeout plus its send tail (up to 16 s) for an answer; counting that tail would leave no room after a long upload that broke, and the fresh budget Failed gives it would buy nothing.

type Conn

type Conn struct {
	net.Conn
	// contains filtered or unexported fields
}

Conn bounds silence on one connection: a Read may wait at most read for the peer to send something, a Write at most write for the peer to take it.

The deadline is set when the call starts, so it bounds a stall, never the length of a transfer, and a caller that is slow to call Read again (a browser reading a download slowly, a paused player) is never cut: the clock runs only while a call waits on the network.

A zero limit leaves that direction unbounded.

func Guard

func Guard(c net.Conn, read, write time.Duration) *Conn

Guard wraps c. read bounds each Read, write each Write.

func (*Conn) AllowOnce

func (c *Conn) AllowOnce(extra time.Duration)

AllowOnce lets the reads wait extra longer until one of them gets something. The answer to an upload may come only once the peer has read what was still in the sockets' buffers when the upload counted as sent (SendTail): FTP's "226" after a STOR against a server that reads slowly.

func (*Conn) Read

func (c *Conn) Read(p []byte) (int, error)

func (*Conn) Write

func (c *Conn) Write(p []byte) (int, error)

type Defaults

type Defaults struct {
	AttemptTimeoutS int
	MaxAttempts     int
	TotalTimeoutS   int
}

Defaults are one driver's defaults for the three settings.

func (Defaults) Fields

func (d Defaults) Fields(t Texts) []storage.Field

Fields are the three settings as descriptor fields, under Advanced, with the driver's defaults and the shared bounds.

type DialFunc

type DialFunc func(ctx context.Context, network, addr string) (net.Conn, error)

DialFunc is net.Dialer.DialContext's shape.

func GuardWrites

func GuardWrites(dial DialFunc, limit time.Duration) DialFunc

GuardWrites makes a send that stopped moving fail. http.Transport has no write timeout: a store that accepted the connection and then stopped reading left the upload blocked in Write for as long as TCP kept the peer alive.

Only writes: an HTTP client's reads are watched per request (AnswerWatch and Reader), because a pooled connection sits in a read between requests.

type Error

type Error struct {
	Kind  Kind
	Limit time.Duration
	Err   error
}

Error is an attempt cut for giving no sign of life. It is a timeout, and a timeout is worth another attempt when the request can be sent again.

func (*Error) Error

func (e *Error) Error() string

func (*Error) RetryableError

func (e *Error) RetryableError() bool

RetryableError answers the AWS SDK's first question directly. Without it the SDK's connection check finds the url.Error inside, looks only at what that wraps — "use of closed network connection", from the watch closing it — and calls the attempt final.

func (*Error) Temporary

func (e *Error) Temporary() bool

func (*Error) Timeout

func (e *Error) Timeout() bool

func (*Error) Unwrap

func (e *Error) Unwrap() error

func (*Error) What

func (e *Error) What() string

What says what ran out, in words, with the limit.

type FieldText

type FieldText struct {
	Label, I18nKey    string
	Help, HelpI18nKey string
}

FieldText is one setting's label and help, in English, with the catalogue keys the surfaces translate them by.

⚠ The key fields are spelled I18nKey / HelpI18nKey, like storage.Field's, and this file is named descriptor.go, on purpose: web/tests/i18n/ storageDescriptorKeys.test.ts finds every key by that spelling in every descriptor.go under internal/storage and fails when the admin catalogue has no English or Turkish for it. Another spelling, or another file name, and a key missing from the catalogue ships as English on a Turkish screen.

type Kind

type Kind int

Kind is what a silence limit was waiting for when it ran out.

const (
	// Answer: the request was sent and no answer started (AnswerWatch).
	Answer Kind = iota
	// Body: an answer started and then stopped arriving (Reader).
	Body
	// Silent: a connection read got nothing from the peer (Conn) — no answer
	// to a command, or a transfer that stopped.
	Silent
	// Send: the peer stopped taking an upload (a watched call that sends,
	// Activity.Watch).
	Send
)

type One added in v0.50.0

type One[S Session] struct {
	// contains filtered or unexported fields
}

One is a driver's one shared session to a server that can go silent (the SFTP and SMB drivers). Current hands out the live session and dials a new one when there is none; Forget retires a session that stopped answering, so the next call dials again; Take empties it for shutdown.

⚠ One caller dials at a time and the others wait for it - or leave as soon as their own context ends, so a dial that hangs on a dead server holds nobody but the caller who started it.

func (*One[S]) Current added in v0.50.0

func (o *One[S]) Current(ctx context.Context, dial func(context.Context) (S, error)) (S, error)

Current returns the live session, dialing one with dial when there is none.

func (*One[S]) Forget added in v0.50.0

func (o *One[S]) Forget(s S)

Forget lets go of s if it is the session held; the caller closes s.

func (*One[S]) Live added in v0.50.0

func (o *One[S]) Live() (S, bool)

Live returns the session held, when there is one and it is not dead.

func (*One[S]) Take added in v0.50.0

func (o *One[S]) Take() (S, bool)

Take empties the holder and returns what it held (shutdown).

type Policy

type Policy struct {
	MaxAttempts    int
	AttemptTimeout time.Duration
	TotalTimeout   time.Duration
}

Policy is one storage's retry and timeout settings.

  • AttemptTimeout: how long one attempt may wait for a sign of life — connecting, the answer to a request, each next piece of a transfer. Never a limit on a transfer that keeps moving.
  • MaxAttempts: how many times a request is tried in all when the failure can pass. A refusal is never retried.
  • TotalTimeout: no new attempt starts unless a store that is still down could show it within this long of the first (see Clock.Room). An attempt under way is not cut.

func (Policy) Do

func (p Policy) Do(ctx context.Context, retry func(error) bool, attempt func(context.Context) error) (*Clock, error)

Do runs attempt until it succeeds, fails for good, or the budget has no room for another try. retry says whether a failed attempt may be sent again — the caller knows whether the request can be repeated (a refusal cannot change, an upload that started sending cannot be replayed); only then do MaxAttempts and the budget decide. The pause between attempts is an exponential backoff with jitter, shortened so the next attempt still fits.

The clock is returned so the caller can report the attempts and the time.

func (Policy) Reason

func (p Policy) Reason(err error) (string, bool)

Reason decides whether err means the store is down — as opposed to a refusal (403, a wrong password, a missing folder, a bad certificate), which is the store answering — and says what happened in words a person can act on, naming the limit that bounded it.

func (Policy) SendStall

func (p Policy) SendStall() time.Duration

SendStall is how long a send may go without moving.

func (Policy) Unavailable

func (p Policy) Unavailable(what string, c *Clock, err error) error

Unavailable wraps err as an UnavailableError when it means the store is down, and returns it unchanged otherwise. what names the store ("webdav server https://dav.example.com"); c is the operation's clock.

type Reader

type Reader struct {
	// contains filtered or unexported fields
}

Reader bounds a stalled answer body. Its timer runs only while a Read is waiting for the network, so a caller that reads slowly is never cut; a store that stops sending mid-listing or mid-download is.

func NewReader

func NewReader(rc io.ReadCloser, limit time.Duration) *Reader

NewReader wraps an answer body.

func (*Reader) Close

func (r *Reader) Close() error

func (*Reader) Read

func (r *Reader) Read(p []byte) (int, error)

type Session added in v0.50.0

type Session interface {
	comparable
	Dead() bool
}

Session is what One holds: a driver's session to its server, which knows when it has been cut.

type Settings

type Settings struct {
	AttemptTimeout any
	MaxAttempts    any
	TotalTimeout   any
}

Settings are the three raw values from a storage's config.

⚠ The driver reads them itself — `Settings{AttemptTimeout: cfg["attempt_timeout_s"], …}` — rather than handing its config map here: the descriptor test (storage/descriptor_drivers_test.go) parses each driver's source for the keys it reads, and a read hidden in this package would look like a declared key nothing reads.

func (Settings) Policy

func (s Settings) Policy(def Defaults) Policy

Policy reads the settings. A missing, empty, zero or unreadable value means the default; a value past the upper bound is brought down to it (the form enforces the same bounds, the CLI and the API do not).

type Texts

type Texts struct {
	AttemptTimeout, MaxAttempts, TotalTimeout FieldText
}

Texts are the three settings' texts.

func ServerTexts

func ServerTexts(attempt FieldText) Texts

ServerTexts are the texts every driver but S3 shows (S3's name its own store-side waits, see drivers/s3/descriptor.go). attempt is the attempt timeout's: AttemptText, or a driver's own when some of its requests wait longer for the server's answer (WebDAV's copies and moves).

type UnavailableError

type UnavailableError struct {
	What     string
	Reason   string
	Attempts int
	Took     time.Duration
	// Err is the failure, for errors.Is/As. Cause, when set, is what the
	// message prints after the colon instead (the S3 driver strips its own
	// wrapper there).
	Err   error
	Cause error
}

UnavailableError is an operation that failed because the store could not be reached or would not answer. It says so first, then the underlying text:

webdav server https://dav.example.com is unavailable: the connection was refused (3 attempts in 14.8s): …

It matches storage.ErrUnavailable, which the HTTP layer answers with 503.

func (*UnavailableError) Error

func (e *UnavailableError) Error() string

func (*UnavailableError) Is

func (e *UnavailableError) Is(target error) bool

func (*UnavailableError) Unwrap

func (e *UnavailableError) Unwrap() error

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL