mirror of
https://github.com/R0m1k3/Socialflow.git
synced 2026-10-23 15:08:13 +02:00
Publications en échec : nouvel essai espacé au lieu de chaque minute
Une publication en échec gardait published_at à NULL et était renvoyée à chaque passage du planificateur, soit toutes les minutes, sans fin. Chaque essai multipliait les appels à l'API Graph, jusqu'à épuiser le quota de l'application (code 4) puis déclencher le blocage anti-spam de Meta (code 368). - scheduled_posts : colonnes attempts et next_attempt_at ; - délai croissant entre les essais (5 min → 3 h), plus long après une limite de débit (30 min → 4 h), abandon après 5 tentatives ; - après une limite de débit, les autres publications de la même plateforme sont reportées au lieu d'aggraver le blocage ; - reprogrammer une publication remet le compteur à zéro ; - l'historique indique la prochaine tentative ou l'abandon. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_018jcWbATGHxatn4V7qS3mr3
This commit is contained in:
9 files changed
+211
-9
No files matched your search
@@ -233,6 +233,15 @@ export default function History() {
|
|||||||
{scheduledPost.error && (
|
{scheduledPost.error && (
|
||||||
<p className="mt-2 rounded-md bg-destructive/10 px-2.5 py-1.5 text-xs text-destructive">{scheduledPost.error}</p>
|
<p className="mt-2 rounded-md bg-destructive/10 px-2.5 py-1.5 text-xs text-destructive">{scheduledPost.error}</p>
|
||||||
)}
|
)}
|
||||||
|
{scheduledPost.error && !scheduledPost.publishedAt && (
|
||||||
|
<p className="mt-1 text-xs text-muted-foreground">
|
||||||
|
{scheduledPost.nextAttemptAt
|
||||||
|
? `Nouvelle tentative automatique le ${format(new Date(scheduledPost.nextAttemptAt), "d MMM yyyy 'à' HH:mm", { locale: fr })}`
|
||||||
|
: scheduledPost.attempts > 0
|
||||||
|
? `Abandon après ${scheduledPost.attempts} tentatives : reprogrammez la publication pour réessayer.`
|
||||||
|
: null}
|
||||||
|
</p>
|
||||||
|
)}
|
||||||
</div>
|
</div>
|
||||||
<Button
|
<Button
|
||||||
variant="ghost"
|
variant="ghost"
|
||||||
|
|||||||
+12
-2
@@ -907,11 +907,21 @@ impl Db {
|
|||||||
}
|
}
|
||||||
|
|
||||||
/// Publications dues et pas encore traitées.
|
/// Publications dues et pas encore traitées.
|
||||||
pub async fn get_pending_scheduled_posts(&self) -> Result<Vec<ScheduledPost>, sqlx::Error> {
|
/// Publications dues, hors échecs en attente de leur prochaine tentative
|
||||||
|
/// et hors publications abandonnées après `max_attempts` échecs.
|
||||||
|
pub async fn get_pending_scheduled_posts(
|
||||||
|
&self,
|
||||||
|
max_attempts: i32,
|
||||||
|
) -> Result<Vec<ScheduledPost>, sqlx::Error> {
|
||||||
sqlx::query_as(
|
sqlx::query_as(
|
||||||
"SELECT * FROM scheduled_posts WHERE published_at IS NULL AND scheduled_at <= $1",
|
r#"SELECT * FROM scheduled_posts
|
||||||
|
WHERE published_at IS NULL AND scheduled_at <= $1
|
||||||
|
AND (next_attempt_at IS NULL OR next_attempt_at <= $1)
|
||||||
|
AND attempts < $2
|
||||||
|
ORDER BY scheduled_at"#,
|
||||||
)
|
)
|
||||||
.bind(Ts::now())
|
.bind(Ts::now())
|
||||||
|
.bind(max_attempts)
|
||||||
.fetch_all(&self.pool)
|
.fetch_all(&self.pool)
|
||||||
.await
|
.await
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -175,6 +175,10 @@ pub struct ScheduledPost {
|
|||||||
pub error: Option<String>,
|
pub error: Option<String>,
|
||||||
pub publish_id: Option<String>,
|
pub publish_id: Option<String>,
|
||||||
pub publish_status: Option<String>,
|
pub publish_status: Option<String>,
|
||||||
|
/// Tentatives de publication en échec.
|
||||||
|
pub attempts: i32,
|
||||||
|
/// Prochaine tentative après un échec (`None` : immédiate ou abandonnée).
|
||||||
|
pub next_attempt_at: Option<Ts>,
|
||||||
}
|
}
|
||||||
|
|
||||||
#[derive(Debug, Clone, Serialize, FromRow)]
|
#[derive(Debug, Clone, Serialize, FromRow)]
|
||||||
|
|||||||
@@ -251,6 +251,9 @@ const COLUMNS: &[&str] = &[
|
|||||||
r#"ALTER TABLE "scheduled_posts"
|
r#"ALTER TABLE "scheduled_posts"
|
||||||
ADD COLUMN IF NOT EXISTS "publish_id" text,
|
ADD COLUMN IF NOT EXISTS "publish_id" text,
|
||||||
ADD COLUMN IF NOT EXISTS "publish_status" text"#,
|
ADD COLUMN IF NOT EXISTS "publish_status" text"#,
|
||||||
|
r#"ALTER TABLE "scheduled_posts"
|
||||||
|
ADD COLUMN IF NOT EXISTS "attempts" integer NOT NULL DEFAULT 0,
|
||||||
|
ADD COLUMN IF NOT EXISTS "next_attempt_at" timestamp"#,
|
||||||
];
|
];
|
||||||
|
|
||||||
/// Index de performance (Postgres n'indexe pas les clés étrangères seul).
|
/// Index de performance (Postgres n'indexe pas les clés étrangères seul).
|
||||||
|
|||||||
@@ -214,6 +214,8 @@ mod tests {
|
|||||||
error: error.map(str::to_string),
|
error: error.map(str::to_string),
|
||||||
publish_id: None,
|
publish_id: None,
|
||||||
publish_status: None,
|
publish_status: None,
|
||||||
|
attempts: 0,
|
||||||
|
next_attempt_at: None,
|
||||||
},
|
},
|
||||||
post: ScheduledPostPost {
|
post: ScheduledPostPost {
|
||||||
id: "p".into(),
|
id: "p".into(),
|
||||||
|
|||||||
@@ -160,7 +160,12 @@ async fn update_scheduled(
|
|||||||
let mut patch = Patch::new();
|
let mut patch = Patch::new();
|
||||||
if let Some(value) = body.get("scheduledAt").filter(|v| is_truthy(Some(v))) {
|
if let Some(value) = body.get("scheduledAt").filter(|v| is_truthy(Some(v))) {
|
||||||
let date = js_date(value).ok_or_else(|| AppError::internal(FAILED))?;
|
let date = js_date(value).ok_or_else(|| AppError::internal(FAILED))?;
|
||||||
patch = patch.set("scheduled_at", date);
|
// Reprogrammer relance les publications en échec ou abandonnées
|
||||||
|
patch = patch
|
||||||
|
.set("scheduled_at", date)
|
||||||
|
.set("attempts", 0)
|
||||||
|
.set("next_attempt_at", None::<Ts>)
|
||||||
|
.set_null("error");
|
||||||
state
|
state
|
||||||
.db
|
.db
|
||||||
.update_post(&scheduled.post_id, Patch::new().set("scheduled_for", date))
|
.update_post(&scheduled.post_id, Patch::new().set("scheduled_for", date))
|
||||||
|
|||||||
@@ -869,7 +869,14 @@ async fn update_post_inner(state: &AppState, id: &str, body: &Value) -> Outcome
|
|||||||
if sp.published_at.is_none() {
|
if sp.published_at.is_none() {
|
||||||
state
|
state
|
||||||
.db
|
.db
|
||||||
.update_scheduled_post(&sp.id, Patch::new().set("scheduled_at", new_date))
|
.update_scheduled_post(
|
||||||
|
&sp.id,
|
||||||
|
Patch::new()
|
||||||
|
.set("scheduled_at", new_date)
|
||||||
|
.set("attempts", 0)
|
||||||
|
.set("next_attempt_at", None::<Ts>)
|
||||||
|
.set_null("error"),
|
||||||
|
)
|
||||||
.await?;
|
.await?;
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -1,9 +1,13 @@
|
|||||||
//! Publication des posts programmés : toutes les minutes, les publications
|
//! Publication des posts programmés : toutes les minutes, les publications
|
||||||
//! dues sont envoyées sur Facebook ou TikTok.
|
//! dues sont envoyées sur Facebook ou TikTok.
|
||||||
|
|
||||||
|
use std::collections::HashMap;
|
||||||
use std::sync::atomic::AtomicBool;
|
use std::sync::atomic::AtomicBool;
|
||||||
use std::sync::Arc;
|
use std::sync::Arc;
|
||||||
|
|
||||||
|
use once_cell::sync::Lazy;
|
||||||
|
use regex::Regex;
|
||||||
|
|
||||||
use crate::db::{
|
use crate::db::{
|
||||||
Media, MediaType, Patch, Platform, Post, PostStatus, PostType, ScheduledPost, SocialPage,
|
Media, MediaType, Patch, Platform, Post, PostStatus, PostType, ScheduledPost, SocialPage,
|
||||||
};
|
};
|
||||||
@@ -29,9 +33,69 @@ pub fn start(state: AppState) {
|
|||||||
tracing::info!("Scheduler service started");
|
tracing::info!("Scheduler service started");
|
||||||
}
|
}
|
||||||
|
|
||||||
|
/// Nombre de tentatives en échec après lequel une publication est abandonnée
|
||||||
|
/// (il faut alors la reprogrammer).
|
||||||
|
pub const MAX_PUBLISH_ATTEMPTS: i32 = 5;
|
||||||
|
|
||||||
|
/// Délai avant la tentative suivante, selon le nombre d'échecs déjà subis.
|
||||||
|
/// Une limite de débit (quota d'appels, blocage anti-spam) appelle des délais
|
||||||
|
/// plus longs : réessayer tout de suite prolonge le blocage.
|
||||||
|
fn retry_delay(attempts: i32, rate_limited: bool) -> chrono::Duration {
|
||||||
|
const MINUTES: [i64; 4] = [5, 15, 60, 180];
|
||||||
|
const RATE_LIMITED_MINUTES: [i64; 4] = [30, 60, 120, 240];
|
||||||
|
let table = if rate_limited {
|
||||||
|
&RATE_LIMITED_MINUTES
|
||||||
|
} else {
|
||||||
|
&MINUTES
|
||||||
|
};
|
||||||
|
let index = usize::try_from(attempts - 1)
|
||||||
|
.unwrap_or(0)
|
||||||
|
.min(table.len() - 1);
|
||||||
|
chrono::Duration::minutes(table[index])
|
||||||
|
}
|
||||||
|
|
||||||
|
/// Code d'erreur Graph repris dans le message (`... (code: 4)`).
|
||||||
|
fn graph_error_code(message: &str) -> Option<i64> {
|
||||||
|
static CODE: Lazy<Regex> = Lazy::new(|| Regex::new(r"\(code: (\d+)").unwrap());
|
||||||
|
CODE.captures(message)?.get(1)?.as_str().parse().ok()
|
||||||
|
}
|
||||||
|
|
||||||
|
/// Erreur due à une limite de débit : quotas de l'application (4), de
|
||||||
|
/// l'utilisateur (17), de la page (32, 613, 80001…) ou blocage temporaire
|
||||||
|
/// anti-spam (368), ou `rate_limit_exceeded` côté TikTok.
|
||||||
|
fn is_rate_limited(message: &str) -> bool {
|
||||||
|
if let Some(code) = graph_error_code(message) {
|
||||||
|
if matches!(code, 4 | 17 | 32 | 368 | 613 | 80001..=80014) {
|
||||||
|
return true;
|
||||||
|
}
|
||||||
|
}
|
||||||
|
let lower = message.to_lowercase();
|
||||||
|
lower.contains("rate_limit") || lower.contains("request limit")
|
||||||
|
}
|
||||||
|
|
||||||
|
async fn page_platform(state: &AppState, page_id: &str) -> Option<Platform> {
|
||||||
|
match state.db.get_social_page(page_id).await {
|
||||||
|
Ok(page) => page.map(|p| p.platform),
|
||||||
|
Err(error) => {
|
||||||
|
tracing::warn!("[Scheduler] Page {page_id} illisible : {error}");
|
||||||
|
None
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
/// Traite les publications dues (une erreur n'interrompt pas les suivantes).
|
/// Traite les publications dues (une erreur n'interrompt pas les suivantes).
|
||||||
|
///
|
||||||
|
/// Un échec n'est retenté qu'après un délai croissant, et au plus
|
||||||
|
/// `MAX_PUBLISH_ATTEMPTS` fois : sans cela, chaque publication en échec était
|
||||||
|
/// renvoyée toutes les minutes et épuisait le quota d'appels de l'application.
|
||||||
|
/// Après une limite de débit, les autres publications de la même plateforme
|
||||||
|
/// sont reportées au lieu d'aggraver le blocage.
|
||||||
pub async fn process_pending_posts(state: &AppState) {
|
pub async fn process_pending_posts(state: &AppState) {
|
||||||
let pending = match state.db.get_pending_scheduled_posts().await {
|
let pending = match state
|
||||||
|
.db
|
||||||
|
.get_pending_scheduled_posts(MAX_PUBLISH_ATTEMPTS)
|
||||||
|
.await
|
||||||
|
{
|
||||||
Ok(pending) => pending,
|
Ok(pending) => pending,
|
||||||
Err(error) => {
|
Err(error) => {
|
||||||
tracing::error!("Error processing pending posts: {error}");
|
tracing::error!("Error processing pending posts: {error}");
|
||||||
@@ -39,14 +103,63 @@ pub async fn process_pending_posts(state: &AppState) {
|
|||||||
}
|
}
|
||||||
};
|
};
|
||||||
|
|
||||||
|
let mut cooldowns: HashMap<Platform, Ts> = HashMap::new();
|
||||||
|
|
||||||
for scheduled_post in pending {
|
for scheduled_post in pending {
|
||||||
|
if !cooldowns.is_empty() {
|
||||||
|
let platform = page_platform(state, &scheduled_post.page_id).await;
|
||||||
|
if let Some(until) = platform.and_then(|p| cooldowns.get(&p)).copied() {
|
||||||
|
tracing::info!(
|
||||||
|
"[Scheduler] Post {} reporté au {until} (limite de débit atteinte)",
|
||||||
|
scheduled_post.id
|
||||||
|
);
|
||||||
|
let deferred = state
|
||||||
|
.db
|
||||||
|
.update_scheduled_post(
|
||||||
|
&scheduled_post.id,
|
||||||
|
Patch::new().set("next_attempt_at", until),
|
||||||
|
)
|
||||||
|
.await;
|
||||||
|
if let Err(error) = deferred {
|
||||||
|
tracing::error!("Error processing pending posts: {error}");
|
||||||
|
return;
|
||||||
|
}
|
||||||
|
continue;
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
if let Err(error) = process_one(state, &scheduled_post).await {
|
if let Err(error) = process_one(state, &scheduled_post).await {
|
||||||
tracing::error!("Error publishing post {}: {error:#}", scheduled_post.id);
|
let message = error_message(&error);
|
||||||
|
let rate_limited = is_rate_limited(&message);
|
||||||
|
let attempts = scheduled_post.attempts + 1;
|
||||||
|
let retry_at = Ts::now().add(retry_delay(attempts, rate_limited));
|
||||||
|
let next_attempt_at = (attempts < MAX_PUBLISH_ATTEMPTS).then_some(retry_at);
|
||||||
|
|
||||||
|
match next_attempt_at {
|
||||||
|
Some(at) => tracing::error!(
|
||||||
|
"Error publishing post {} (tentative {attempts}/{MAX_PUBLISH_ATTEMPTS}, nouvel essai le {at}): {error:#}",
|
||||||
|
scheduled_post.id
|
||||||
|
),
|
||||||
|
None => tracing::error!(
|
||||||
|
"Error publishing post {} (abandon après {attempts} tentatives): {error:#}",
|
||||||
|
scheduled_post.id
|
||||||
|
),
|
||||||
|
}
|
||||||
|
|
||||||
|
if rate_limited {
|
||||||
|
if let Some(platform) = page_platform(state, &scheduled_post.page_id).await {
|
||||||
|
cooldowns.insert(platform, retry_at);
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
let recorded = state
|
let recorded = state
|
||||||
.db
|
.db
|
||||||
.update_scheduled_post(
|
.update_scheduled_post(
|
||||||
&scheduled_post.id,
|
&scheduled_post.id,
|
||||||
Patch::new().set("error", error_message(&error)),
|
Patch::new()
|
||||||
|
.set("error", message)
|
||||||
|
.set("attempts", attempts)
|
||||||
|
.set("next_attempt_at", next_attempt_at),
|
||||||
)
|
)
|
||||||
.await;
|
.await;
|
||||||
if let Err(error) = recorded {
|
if let Err(error) = recorded {
|
||||||
@@ -136,7 +249,8 @@ async fn publish_post(state: &AppState, scheduled_post: &ScheduledPost) -> anyho
|
|||||||
Patch::new()
|
Patch::new()
|
||||||
.set("published_at", Ts::now())
|
.set("published_at", Ts::now())
|
||||||
.set("external_post_id", external_post_id)
|
.set("external_post_id", external_post_id)
|
||||||
.set_null("error"),
|
.set_null("error")
|
||||||
|
.set("next_attempt_at", None::<Ts>),
|
||||||
)
|
)
|
||||||
.await?;
|
.await?;
|
||||||
}
|
}
|
||||||
@@ -154,7 +268,8 @@ async fn publish_post(state: &AppState, scheduled_post: &ScheduledPost) -> anyho
|
|||||||
.set("published_at", Ts::now())
|
.set("published_at", Ts::now())
|
||||||
.set("publish_id", publish_id)
|
.set("publish_id", publish_id)
|
||||||
.set("publish_status", "PROCESSING_UPLOAD")
|
.set("publish_status", "PROCESSING_UPLOAD")
|
||||||
.set_null("error"),
|
.set_null("error")
|
||||||
|
.set("next_attempt_at", None::<Ts>),
|
||||||
)
|
)
|
||||||
.await?;
|
.await?;
|
||||||
}
|
}
|
||||||
@@ -254,3 +369,43 @@ async fn publish_to_tiktok(
|
|||||||
|
|
||||||
tiktok::publish_video_from_buffer(state, page, buffer, &post.content).await
|
tiktok::publish_video_from_buffer(state, page, buffer, &post.content).await
|
||||||
}
|
}
|
||||||
|
|
||||||
|
#[cfg(test)]
|
||||||
|
mod tests {
|
||||||
|
use super::*;
|
||||||
|
|
||||||
|
#[test]
|
||||||
|
fn detects_facebook_rate_limits() {
|
||||||
|
assert!(is_rate_limited(
|
||||||
|
"Facebook Story API error (START phase): (#4) Application request limit reached (code: 4)"
|
||||||
|
));
|
||||||
|
assert!(is_rate_limited(
|
||||||
|
"Facebook Reel START error: Nous limitons le nombre de fois que vous pouvez publier (code: 368, subcode: 1390008)"
|
||||||
|
));
|
||||||
|
assert!(is_rate_limited(
|
||||||
|
"Facebook API error: Page request limit reached (code: 32)"
|
||||||
|
));
|
||||||
|
assert!(is_rate_limited("TikTok error: rate_limit_exceeded"));
|
||||||
|
}
|
||||||
|
|
||||||
|
#[test]
|
||||||
|
fn other_errors_are_not_rate_limits() {
|
||||||
|
assert!(!is_rate_limited(
|
||||||
|
"Facebook API error: Error validating access token (code: 190)"
|
||||||
|
));
|
||||||
|
assert!(!is_rate_limited(
|
||||||
|
"Facebook API error: Invalid parameter (code: 100)"
|
||||||
|
));
|
||||||
|
assert!(!is_rate_limited("Post or page not found"));
|
||||||
|
}
|
||||||
|
|
||||||
|
#[test]
|
||||||
|
fn retry_delays_grow_and_stay_bounded() {
|
||||||
|
assert_eq!(retry_delay(1, false), chrono::Duration::minutes(5));
|
||||||
|
assert_eq!(retry_delay(2, false), chrono::Duration::minutes(15));
|
||||||
|
assert_eq!(retry_delay(9, false), chrono::Duration::minutes(180));
|
||||||
|
assert_eq!(retry_delay(1, true), chrono::Duration::minutes(30));
|
||||||
|
assert_eq!(retry_delay(4, true), chrono::Duration::minutes(240));
|
||||||
|
assert_eq!(retry_delay(0, true), chrono::Duration::minutes(30));
|
||||||
|
}
|
||||||
|
}
|
||||||
@@ -230,6 +230,11 @@ export const scheduledPosts = pgTable("scheduled_posts", {
|
|||||||
// post_id définitif n'est connu qu'après interrogation du statut.
|
// post_id définitif n'est connu qu'après interrogation du statut.
|
||||||
publishId: text("publish_id"),
|
publishId: text("publish_id"),
|
||||||
publishStatus: text("publish_status"),
|
publishStatus: text("publish_status"),
|
||||||
|
|
||||||
|
// Échecs de publication : nouvel essai à `nextAttemptAt` (délai croissant),
|
||||||
|
// abandon après quelques tentatives (il faut alors reprogrammer).
|
||||||
|
attempts: integer("attempts").notNull().default(0),
|
||||||
|
nextAttemptAt: timestamp("next_attempt_at"),
|
||||||
});
|
});
|
||||||
|
|
||||||
export const aiGenerations = pgTable("ai_generations", {
|
export const aiGenerations = pgTable("ai_generations", {
|
||||||
@@ -454,6 +459,8 @@ export const insertScheduledPostSchema = createInsertSchema(scheduledPosts).omit
|
|||||||
publishedAt: true,
|
publishedAt: true,
|
||||||
externalPostId: true,
|
externalPostId: true,
|
||||||
error: true,
|
error: true,
|
||||||
|
attempts: true,
|
||||||
|
nextAttemptAt: true,
|
||||||
});
|
});
|
||||||
|
|
||||||
export const insertAiGenerationSchema = createInsertSchema(aiGenerations).omit({
|
export const insertAiGenerationSchema = createInsertSchema(aiGenerations).omit({
|
||||||
|
|||||||
Reference in new issue
Block a user