diff --git a/src/env.go b/src/env.go index f3c82b9..0cabc93 100644 --- a/src/env.go +++ b/src/env.go @@ -24,8 +24,26 @@ 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"` + // 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/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..6306254 100644 --- a/src/gows/manager.go +++ b/src/gows/manager.go @@ -85,13 +85,52 @@ 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) { + // 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 } +// 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 +} + +// 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 74180d5..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,35 +45,77 @@ 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 + // 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 { - 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, @@ -80,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 4680960..1892771 100644 --- a/src/main.go +++ b/src/main.go @@ -138,6 +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, 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) 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: