From d812770e019ce7f1aa9fc3179b99db5675461fd4 Mon Sep 17 00:00:00 2001 From: 9bany Date: Wed, 14 Jan 2026 23:26:49 +0700 Subject: [PATCH] fix ns --- crates/core/backend.rs | 15 +++++++++++++++ crates/core/worker.rs | 10 ++++++++++ crates/mysql/lib.rs | 5 +++++ crates/postgres/lib.rs | 5 +++++ crates/redis/lib.rs | 5 +++++ crates/sqlite/lib.rs | 5 +++++ 6 files changed, 45 insertions(+) diff --git a/crates/core/backend.rs b/crates/core/backend.rs index e11eb77..6f50e9a 100644 --- a/crates/core/backend.rs +++ b/crates/core/backend.rs @@ -161,6 +161,17 @@ pub trait Backend: Send + Sync { /// This is useful for admin/monitoring tools to discover all active namespaces. async fn list_namespaces(&self) -> Result>; + /// Initialize the namespace (create required tables/structures). + /// + /// This is idempotent - safe to call multiple times. SQL backends will + /// create tables if they don't exist. Redis and other backends that + /// don't require initialization will do nothing. + async fn init_namespace(&self, namespace: &str) -> Result<()> { + // Default implementation is no-op for backends that don't need initialization + let _ = namespace; + Ok(()) + } + // ========== Concurrency Control ========== /// Set the maximum concurrency for a job type in a namespace. @@ -318,6 +329,10 @@ impl Backend for SharedBackend { self.inner.list_namespaces().await } + async fn init_namespace(&self, namespace: &str) -> Result<()> { + self.inner.init_namespace(namespace).await + } + async fn set_job_concurrency(&self, ns: &str, job_name: &str, max: usize) -> Result<()> { self.inner.set_job_concurrency(ns, job_name, max).await } diff --git a/crates/core/worker.rs b/crates/core/worker.rs index fea4907..88eeaa5 100644 --- a/crates/core/worker.rs +++ b/crates/core/worker.rs @@ -123,6 +123,16 @@ where .take() .ok_or_else(|| WgError::Config("Backend not configured".to_string()))?; + // Initialize namespace if needed (idempotent for SQL backends) + if let Err(e) = backend.init_namespace(&self.config.namespace).await { + tracing::warn!( + error = %e, + namespace = %self.config.namespace, + "Failed to initialize namespace, continuing anyway" + ); + // Continue anyway - might already be initialized or backend doesn't need it + } + self.running.store(true, Ordering::SeqCst); let mut tasks = JoinSet::new(); diff --git a/crates/mysql/lib.rs b/crates/mysql/lib.rs index ca156d9..1346020 100644 --- a/crates/mysql/lib.rs +++ b/crates/mysql/lib.rs @@ -682,6 +682,11 @@ impl Backend for MySqlBackend { Ok(result) } + async fn init_namespace(&self, namespace: &str) -> Result<()> { + // Delegate to the struct's init_namespace method + ::init_namespace(self, namespace).await + } + // ========== Concurrency Control ========== async fn set_job_concurrency(&self, ns: &str, job_name: &str, max: usize) -> Result<()> { diff --git a/crates/postgres/lib.rs b/crates/postgres/lib.rs index 95f0e8a..e2044f5 100644 --- a/crates/postgres/lib.rs +++ b/crates/postgres/lib.rs @@ -698,6 +698,11 @@ impl Backend for PostgresBackend { Ok(result) } + async fn init_namespace(&self, namespace: &str) -> Result<()> { + // Delegate to the struct's init_namespace method + ::init_namespace(self, namespace).await + } + // ========== Concurrency Control ========== async fn set_job_concurrency(&self, ns: &str, job_name: &str, max: usize) -> Result<()> { diff --git a/crates/redis/lib.rs b/crates/redis/lib.rs index e675e19..922aa68 100644 --- a/crates/redis/lib.rs +++ b/crates/redis/lib.rs @@ -582,6 +582,11 @@ impl Backend for RedisBackend { Ok(result) } + async fn init_namespace(&self, _namespace: &str) -> Result<()> { + // Redis doesn't require namespace initialization - keys are created on demand + Ok(()) + } + // ========== Concurrency Control ========== async fn set_job_concurrency(&self, ns: &str, job_name: &str, max: usize) -> Result<()> { diff --git a/crates/sqlite/lib.rs b/crates/sqlite/lib.rs index fdc193d..6513bd7 100644 --- a/crates/sqlite/lib.rs +++ b/crates/sqlite/lib.rs @@ -709,6 +709,11 @@ impl Backend for SqliteBackend { Ok(result) } + async fn init_namespace(&self, namespace: &str) -> Result<()> { + // Delegate to the struct's init_namespace method + ::init_namespace(self, namespace).await + } + // ========== Concurrency Control ========== async fn set_job_concurrency(&self, ns: &str, job_name: &str, max: usize) -> Result<()> {