Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
15 changes: 15 additions & 0 deletions crates/core/backend.rs
Original file line number Diff line number Diff line change
Expand Up @@ -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<Vec<String>>;

/// 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.
Expand Down Expand Up @@ -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
}
Expand Down
10 changes: 10 additions & 0 deletions crates/core/worker.rs
Original file line number Diff line number Diff line change
Expand Up @@ -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();
Expand Down
5 changes: 5 additions & 0 deletions crates/mysql/lib.rs
Original file line number Diff line number Diff line change
Expand Up @@ -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
<MySqlBackend>::init_namespace(self, namespace).await
}

// ========== Concurrency Control ==========

async fn set_job_concurrency(&self, ns: &str, job_name: &str, max: usize) -> Result<()> {
Expand Down
5 changes: 5 additions & 0 deletions crates/postgres/lib.rs
Original file line number Diff line number Diff line change
Expand Up @@ -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
<PostgresBackend>::init_namespace(self, namespace).await
}

// ========== Concurrency Control ==========

async fn set_job_concurrency(&self, ns: &str, job_name: &str, max: usize) -> Result<()> {
Expand Down
5 changes: 5 additions & 0 deletions crates/redis/lib.rs
Original file line number Diff line number Diff line change
Expand Up @@ -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<()> {
Expand Down
5 changes: 5 additions & 0 deletions crates/sqlite/lib.rs
Original file line number Diff line number Diff line change
Expand Up @@ -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
<SqliteBackend>::init_namespace(self, namespace).await
}

// ========== Concurrency Control ==========

async fn set_job_concurrency(&self, ns: &str, job_name: &str, max: usize) -> Result<()> {
Expand Down