Retry failed status batches instead of failing the whole send

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 {

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
 		}

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) {

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,51 +33,162 @@ 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 == "" {
 		extra.ID = gows.Client.GenerateMessageID()
 	}
 
 	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,
 		Timestamp: time.Now(),
 	}
 	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 <code>" 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()
+	}
+}

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)

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: