From 78f16fcb69f1e49e6f65babff7608e29fbc6d22f Mon Sep 17 00:00:00 2001 From: Elimeshi1 Date: Wed, 5 Aug 2026 19:13:17 +0300 Subject: [PATCH 1/2] Fix large status@broadcast sends timing out Sending a status to a large contact list failed with "timed out waiting for message send response". Shrinking the participants batch from 5000 to 500 did not fix it: it multiplied the number of stanzas by ten and traded the timeouts for "server returned error 429". There were two independent causes. The per-batch ack window. whatsmeow defaults to 75s, which a big status batch can exceed even when it is delivered, so the send is reported as a timeout. Batch size was never the problem: what trips WhatsApp's rate limiting is the number of stanzas, not the participants inside one. Set an explicit timeout (default 180s, WAHA_GOWS_STATUS_BATCH_TIMEOUT) and restore the 5000 default. Database lock contention. The sqlite device store was opened in the default rollback-journal mode, where one writer holds an exclusive lock. During a status broadcast the outgoing send and the flood of incoming decryptions (session/identity/sender-key writes) contend for it, exhaust the 30s busy_timeout and surface as "database is locked" - turning a ~4 minute send into a ~30 minute stall ending in a gRPC DEADLINE_EXCEEDED. Open it with WAL + synchronous=NORMAL + txlock=immediate, mirroring the GContainer store. Also detach the status send from the caller's context. A deadline on the caller side was aborting a send already live on WhatsApp, leaving the later batches undelivered while the caller saw a plain failure and re-sent the whole status - posting it two or three times. Fixes devlikeapro/waha#2096 --- src/env.go | 13 +++++++++++-- src/gows/gows.go | 9 ++++++++- src/gows/manager.go | 13 +++++++++++-- src/gows/status.go | 6 ++++++ src/main.go | 2 ++ src/server/session.go | 12 +++++++++++- 6 files changed, 49 insertions(+), 6 deletions(-) diff --git a/src/env.go b/src/env.go index f3c82b9..9fd4e91 100644 --- a/src/env.go +++ b/src/env.go @@ -24,8 +24,17 @@ func getClientConfig() ClientConfig { type StatusConfig struct { // ParticipantsBatchSize controls how many contacts are included per batch // when sending a status/story to status@broadcast. - // Lowered from 5000 to 500 to avoid gRPC batch timeout errors on large contact lists. - ParticipantsBatchSize int `env:"WAHA_GOWS_STATUS_PARTICIPANTS_BATCH_SIZE" envDefault:"500"` + // Kept large on purpose: what trips WhatsApp's rate limiting is the number of + // stanzas, not the number of participants inside one. Smaller batches turned + // the ack timeouts into "server returned error 429" instead of fixing them. + // Slow acks on a big batch are handled by BatchTimeout below. + ParticipantsBatchSize int `env:"WAHA_GOWS_STATUS_PARTICIPANTS_BATCH_SIZE" envDefault:"5000"` + // BatchTimeout bounds how long we wait for the server to acknowledge one + // batch. whatsmeow defaults to 75s, which is not enough for a large status + // batch and shows up as "timed out waiting for message send response" even + // though the batch is delivered. + // Go duration format: 90s, 180s, 5m. + BatchTimeout time.Duration `env:"WAHA_GOWS_STATUS_BATCH_TIMEOUT" envDefault:"180s"` } func getStatusConfig() StatusConfig { diff --git a/src/gows/gows.go b/src/gows/gows.go index e1e0c44..5081614 100644 --- a/src/gows/gows.go +++ b/src/gows/gows.go @@ -432,7 +432,14 @@ func (gows *GoWS) SendMessage(ctx context.Context, to types.JID, msg *waE2E.Mess if to.User == "status" && to.Server == types.BroadcastServer { // Broadcast messages (Status) - result, err := gows.SendStatusMessage(ctx, to, msg, extra) + // + // A status to a large audience can outlast the caller's deadline. Detach + // from its cancellation so a client-side timeout does not abort a send that + // is already live on WhatsApp: aborting it mid-batch leaves the remaining + // batches undelivered while the caller sees a plain failure and re-sends + // the whole status, posting it more than once. + sendCtx := context.WithoutCancel(ctx) + result, err := gows.SendStatusMessage(sendCtx, to, msg, extra) if err != nil { return nil, err } diff --git a/src/gows/manager.go b/src/gows/manager.go index d2488b9..e466398 100644 --- a/src/gows/manager.go +++ b/src/gows/manager.go @@ -85,13 +85,22 @@ func SetDeviceAndBrowser(device string, browser string) { } // statusParticipantsBatchSize is the number of contacts per batch when sending to status@broadcast. -// Set at startup via SetStatusParticipantsBatchSize; defaults to 500. -var statusParticipantsBatchSize = 500 +// Set at startup via SetStatusParticipantsBatchSize; defaults to 5000. +var statusParticipantsBatchSize = 5000 func SetStatusParticipantsBatchSize(n int) { statusParticipantsBatchSize = n } +// statusBatchTimeout is how long to wait for the server to ack one status batch. +// Set at startup via SetStatusBatchTimeout; defaults to 180s. Zero leaves +// whatsmeow's own default (75s) in place. +var statusBatchTimeout = 180 * time.Second + +func SetStatusBatchTimeout(d time.Duration) { + statusBatchTimeout = d +} + // SetKeepAliveInterval overrides whatsmeow's websocket keepalive ping interval. // Returns the resulting min/max so the caller can log what is in effect. func SetKeepAliveInterval(min time.Duration, max time.Duration) (time.Duration, time.Duration) { diff --git a/src/gows/status.go b/src/gows/status.go index 74180d5..9ea4ef6 100644 --- a/src/gows/status.go +++ b/src/gows/status.go @@ -56,6 +56,12 @@ func (gows *GoWS) SendStatusMessage(ctx context.Context, to types.JID, msg *waE2 for index, participants := range batches { batchExtra := extra batchExtra.Participants = participants + // Give the server room to acknowledge a large batch. Without this the + // whatsmeow default (75s) applies, and a batch that is actually delivered + // is reported as "timed out waiting for message send response". + if statusBatchTimeout > 0 { + batchExtra.Timeout = statusBatchTimeout + } _, err := gows.Client.SendMessage(ctx, to, msg, batchExtra) if err != nil { diff --git a/src/main.go b/src/main.go index 4680960..b88944a 100644 --- a/src/main.go +++ b/src/main.go @@ -138,6 +138,8 @@ func main() { statusCfg := getStatusConfig() log.Infof("Status broadcast participants batch size: %d", statusCfg.ParticipantsBatchSize) gows.SetStatusParticipantsBatchSize(statusCfg.ParticipantsBatchSize) + log.Infof("Status broadcast batch ack timeout: %s", statusCfg.BatchTimeout) + gows.SetStatusBatchTimeout(statusCfg.BatchTimeout) linkPreviewCfg := getLinkPreviewConfig() log.Infof("Link preview fetch timeout: %s", linkPreviewCfg.FetchTimeout) diff --git a/src/server/session.go b/src/server/session.go index da01aad..680b325 100644 --- a/src/server/session.go +++ b/src/server/session.go @@ -30,7 +30,17 @@ func (s *Server) StartSession(ctx context.Context, req *__.StartSessionRequest) case dialect == "sqlite3" || dialect == "sqlite": // busy_timeout to prevent "database is locked" errors // DO NOT add cache=shared, it's not safe - address = req.Config.Store.Address + "?_foreign_keys=on&_busy_timeout=30000" + // + // WAL is required here, not optional. In the default rollback-journal mode a + // single writer takes an exclusive lock that blocks every other reader and + // writer. During a status@broadcast the outgoing send and the flood of + // incoming decryptions (each a session/identity/sender-key write) contend for + // that one lock, exhaust the busy_timeout above and surface as "database is + // locked". WAL lets one writer proceed alongside readers, and + // _txlock=immediate takes the write lock up front so concurrent writers back + // off cleanly instead of deadlocking mid-transaction. + // (Mirrors the PRAGMAs already applied to the GContainer store.) + address = req.Config.Store.Address + "?_foreign_keys=on&_busy_timeout=30000&_journal_mode=WAL&_synchronous=NORMAL&_txlock=immediate" case dialect == "postgres": address = addApplicationName(req.Config.Store.Address, "GOWS") default: From 4a357699236b6c7b31796ce243f23c8a008ebd9e Mon Sep 17 00:00:00 2001 From: Elimeshi1 Date: Wed, 5 Aug 2026 19:16:26 +0300 Subject: [PATCH 2/2] Retry failed status batches instead of failing the whole send Three related changes to how a status broadcast handles a batch that fails. Retry transient failures. An ack timeout or a 429 is a temporary condition, but the loop gave up on the batch immediately. Retry it (default 2 attempts, 5s then 15s). Only those two conditions are retried: any other server error is permanent and would just burn the budget and delay the batches behind it. Retrying is safe because the message ID is generated once for the whole status and reused by every batch and attempt, and WhatsApp deduplicates by (sender, message ID). Deliver best effort. A single failed batch failed the entire call, even when the other batches were already delivered - and a status that reached anyone is live on WhatsApp. The caller read that error as "reached nobody" and sent the status again, posting it two or three times. Now the call fails only when no batch got through, and reports which batch numbers were delivered. Space the batches out. They were sent back to back with no gap; a fixed delay (default 1.5s) keeps the cadence steady rather than bursty. All values are configurable: WAHA_GOWS_STATUS_BATCH_DELAY, WAHA_GOWS_STATUS_BATCH_MAX_RETRIES, WAHA_GOWS_STATUS_BATCH_RETRY_BACKOFF. --- src/env.go | 9 ++++ src/gows/manager.go | 30 +++++++++++ src/gows/status.go | 128 ++++++++++++++++++++++++++++++++++++++++---- src/main.go | 10 +++- 4 files changed, 165 insertions(+), 12 deletions(-) diff --git a/src/env.go b/src/env.go index 9fd4e91..0cabc93 100644 --- a/src/env.go +++ b/src/env.go @@ -35,6 +35,15 @@ type StatusConfig struct { // though the batch is delivered. // Go duration format: 90s, 180s, 5m. BatchTimeout time.Duration `env:"WAHA_GOWS_STATUS_BATCH_TIMEOUT" envDefault:"180s"` + // BatchDelay is the pause between batches, so a large audience goes out at a + // steady cadence instead of as a burst of back-to-back stanzas. + BatchDelay time.Duration `env:"WAHA_GOWS_STATUS_BATCH_DELAY" envDefault:"1500ms"` + // BatchMaxRetries is how many extra attempts a batch gets after a transient + // failure (ack timeout or 429). Zero disables retrying. + BatchMaxRetries int `env:"WAHA_GOWS_STATUS_BATCH_MAX_RETRIES" envDefault:"2"` + // BatchRetryBackoff is the wait before the first retry; it triples on each + // further attempt (5s, 15s, ...). + BatchRetryBackoff time.Duration `env:"WAHA_GOWS_STATUS_BATCH_RETRY_BACKOFF" envDefault:"5s"` } func getStatusConfig() StatusConfig { diff --git a/src/gows/manager.go b/src/gows/manager.go index e466398..6306254 100644 --- a/src/gows/manager.go +++ b/src/gows/manager.go @@ -89,6 +89,11 @@ func SetDeviceAndBrowser(device string, browser string) { var statusParticipantsBatchSize = 5000 func SetStatusParticipantsBatchSize(n int) { + // lo.Chunk requires a positive size, and the batch size is now the only thing + // deciding how a status is split, so refuse a value that would panic. + if n <= 0 { + return + } statusParticipantsBatchSize = n } @@ -101,6 +106,31 @@ func SetStatusBatchTimeout(d time.Duration) { statusBatchTimeout = d } +// statusBatchDelay is the pause between status batches. +// Set at startup via SetStatusBatchDelay; defaults to 1.5s. +var statusBatchDelay = 1500 * time.Millisecond + +func SetStatusBatchDelay(d time.Duration) { + statusBatchDelay = d +} + +// statusBatchMaxRetries is how many extra attempts a status batch gets after a +// transient failure. Set at startup via SetStatusBatchRetry; defaults to 2. +var statusBatchMaxRetries = 2 + +// statusBatchRetryBackoff is the wait before the first retry, tripling on each +// further attempt. Set at startup via SetStatusBatchRetry; defaults to 5s. +var statusBatchRetryBackoff = 5 * time.Second + +func SetStatusBatchRetry(maxRetries int, backoff time.Duration) { + if maxRetries >= 0 { + statusBatchMaxRetries = maxRetries + } + if backoff > 0 { + statusBatchRetryBackoff = backoff + } +} + // SetKeepAliveInterval overrides whatsmeow's websocket keepalive ping interval. // Returns the resulting min/max so the caller can log what is in effect. func SetKeepAliveInterval(min time.Duration, max time.Duration) (time.Duration, time.Duration) { diff --git a/src/gows/status.go b/src/gows/status.go index 9ea4ef6..730552c 100644 --- a/src/gows/status.go +++ b/src/gows/status.go @@ -9,6 +9,7 @@ import ( "go.mau.fi/whatsmeow" "go.mau.fi/whatsmeow/proto/waE2E" "go.mau.fi/whatsmeow/types" + "strings" "time" ) @@ -32,11 +33,11 @@ func (gows *GoWS) SendStatusMessage(ctx context.Context, to types.JID, msg *waE2 return p.Server == types.DefaultUserServer }) + // Always batch by the configured size, including when the caller supplied the + // participant list. Using len(extra.Participants) as the batch size sent an + // explicit list as one single batch - precisely the large send that has to be + // split. participantsBatchSize := statusParticipantsBatchSize - if len(extra.Participants) > 0 { - // If participants are provided, use the batch size of the participants - participantsBatchSize = len(extra.Participants) - } batches := lo.Chunk(validParticipants, participantsBatchSize) if extra.ID == "" { @@ -44,16 +45,30 @@ func (gows *GoWS) SendStatusMessage(ctx context.Context, to types.JID, msg *waE2 } errs := make([]error, 0) + succeeded := 0 + deliveredBatches := make([]int, 0, len(batches)) + failedBatches := make([]int, 0) ignored := len(allParticipants) - len(validParticipants) gows.Log.Infof( - "Sending status message (%s) in %d batches - %d participants in total, %d per batch, %d ignored", + "Sending status message (%s) in %d batches - %d participants in total, %d per batch, %d ignored (timeout=%s, delay=%s, retries=%d)", extra.ID, len(batches), len(validParticipants), participantsBatchSize, ignored, + statusBatchTimeout, + statusBatchDelay, + statusBatchMaxRetries, ) for index, participants := range batches { + // Steady cadence rather than a burst of back-to-back stanzas. Not before + // the first batch, there is nothing to space it from. + if index > 0 { + if err := sleepCtx(ctx, statusBatchDelay); err != nil { + return nil, err + } + } + batchExtra := extra batchExtra.Participants = participants // Give the server room to acknowledge a large batch. Without this the @@ -63,22 +78,44 @@ func (gows *GoWS) SendStatusMessage(ctx context.Context, to types.JID, msg *waE2 batchExtra.Timeout = statusBatchTimeout } - _, err := gows.Client.SendMessage(ctx, to, msg, batchExtra) - if err != nil { - gows.Log.Errorf("Failed to send message (%s) to (batch %d/%d): %v", extra.ID, index+1, len(batches), err) - errs = append(errs, fmt.Errorf("batch %d: %w", index+1, err)) + batchErr := gows.sendStatusBatchWithRetry(ctx, to, msg, batchExtra, index+1, len(batches)) + if batchErr != nil { + gows.Log.Errorf("Failed to send message (%s) to (batch %d/%d): %v", extra.ID, index+1, len(batches), batchErr) + errs = append(errs, fmt.Errorf("batch %d: %w", index+1, batchErr)) + failedBatches = append(failedBatches, index+1) } else { + succeeded++ + deliveredBatches = append(deliveredBatches, index+1) gows.Log.Infof("Sending status message (%s) to %d participants (batch %d/%d) - success", extra.ID, len(participants), index+1, len(batches)) } } - if len(errs) > 0 { + // Report at batch granularity, so a broadcast is never an opaque + // success/failure. Logged before the total-failure return below, so a send + // that reached nobody still records what it attempted. + gows.Log.Infof( + "Status (%s) delivery report: %d/%d batches delivered (ok=%v, failed=%v), %d ignored", + extra.ID, len(deliveredBatches), len(batches), deliveredBatches, failedBatches, ignored, + ) + + // Best effort: fail the call only when nothing got through. A status that + // reached at least one batch is already live on WhatsApp, so reporting a + // partial failure as an error only pushes the caller to send the whole thing + // again - which posts the status a second time. + if succeeded == 0 && len(errs) > 0 { err = errors.Join(errs...) gows.Log.Errorf("Failed to send status message (%s): %v", extra.ID, err) return nil, err } - gows.Log.Infof("Sending status message (%s) - success", extra.ID) + if len(errs) > 0 { + gows.Log.Warnf( + "Status message (%s) partially delivered: %d/%d batches ok", + extra.ID, succeeded, len(batches), + ) + } else { + gows.Log.Infof("Sending status message (%s) - success", extra.ID) + } result := &whatsmeow.SendResponse{ ID: extra.ID, @@ -86,3 +123,72 @@ func (gows *GoWS) SendStatusMessage(ctx context.Context, to types.JID, msg *waE2 } return result, nil } + +// sendStatusBatchWithRetry sends one batch, retrying transient failures with an +// exponential backoff. +// +// Retrying is safe: extra.ID is generated once for the whole status and every +// batch and every attempt reuses it, and WhatsApp deduplicates by (sender, +// message ID). A recipient that already received the batch does not see the +// status twice. +func (gows *GoWS) sendStatusBatchWithRetry( + ctx context.Context, + to types.JID, + msg *waE2E.Message, + batchExtra whatsmeow.SendRequestExtra, + batchNum, batchTotal int, +) error { + var lastErr error + backoff := statusBatchRetryBackoff + for attempt := 0; attempt <= statusBatchMaxRetries; attempt++ { + if attempt > 0 { + gows.Log.Warnf( + "Retrying status batch %d/%d (attempt %d/%d) after %s - previous error: %v", + batchNum, batchTotal, attempt, statusBatchMaxRetries, backoff, lastErr, + ) + if err := sleepCtx(ctx, backoff); err != nil { + return err + } + backoff *= 3 + } + + _, err := gows.Client.SendMessage(ctx, to, msg, batchExtra) + if err == nil { + return nil + } + lastErr = err + if !isRetryableStatusErr(err) { + return err + } + } + return lastErr +} + +// isRetryableStatusErr reports whether a failed status batch is worth sending +// again. Only two conditions are transient: an ack timeout, and an explicit 429 +// rate limit. Any other server error - a permanent rejection, a capping nack - +// would just burn the retry budget and delay the remaining batches. +// +// whatsmeow renders server errors as "server returned error " with the +// code appended as text, so matching the code means matching the suffix; there +// is no structured error to inspect. +func isRetryableStatusErr(err error) bool { + if errors.Is(err, whatsmeow.ErrMessageTimedOut) { + return true + } + return errors.Is(err, whatsmeow.ErrServerReturnedError) && + strings.HasSuffix(err.Error(), " 429") +} + +// sleepCtx waits for d, returning early if the context is cancelled. +func sleepCtx(ctx context.Context, d time.Duration) error { + if d <= 0 { + return nil + } + select { + case <-time.After(d): + return nil + case <-ctx.Done(): + return ctx.Err() + } +} diff --git a/src/main.go b/src/main.go index b88944a..1892771 100644 --- a/src/main.go +++ b/src/main.go @@ -138,8 +138,16 @@ func main() { statusCfg := getStatusConfig() log.Infof("Status broadcast participants batch size: %d", statusCfg.ParticipantsBatchSize) gows.SetStatusParticipantsBatchSize(statusCfg.ParticipantsBatchSize) - log.Infof("Status broadcast batch ack timeout: %s", statusCfg.BatchTimeout) + log.Infof( + "Status broadcast batch ack timeout: %s, delay between batches: %s, retries: %d (backoff %s)", + statusCfg.BatchTimeout, + statusCfg.BatchDelay, + statusCfg.BatchMaxRetries, + statusCfg.BatchRetryBackoff, + ) gows.SetStatusBatchTimeout(statusCfg.BatchTimeout) + gows.SetStatusBatchDelay(statusCfg.BatchDelay) + gows.SetStatusBatchRetry(statusCfg.BatchMaxRetries, statusCfg.BatchRetryBackoff) linkPreviewCfg := getLinkPreviewConfig() log.Infof("Link preview fetch timeout: %s", linkPreviewCfg.FetchTimeout)