Commit

Author:

Hash:

Timestamp:

+97 -104 +/-7 browse

Kevin Schoon [me@kevinschoon.com]

45edeef43c5eb44e0e127d3599beb77e5eac212a

Tue, 07 Jul 2026 09:29:43 +0000 (1 week ago)

flatten all db errors into one ayllu_database::Error enum
1diff --git a/ayllu-build/src/error.rs b/ayllu-build/src/error.rs
2index 2867294..1d815c9 100644
3--- a/ayllu-build/src/error.rs
4+++ b/ayllu-build/src/error.rs
5 @@ -32,10 +32,6 @@ pub enum Error {
6 DuplicateWorkflows,
7 #[error("SQL Operation failed: {0}")]
8 Database(#[from] ayllu_database::Error),
9- #[error("DB Connection: {0}")]
10- DatabaseConnection(#[from] ayllu_database::ConnectionError),
11- #[error("Blob error: {0}")]
12- OutputCopyFailure(#[from] ayllu_database::RawError),
13 // general io error during execution
14 #[error("Some other IO error: {0}")]
15 Io(#[from] std::io::Error),
16 diff --git a/ayllu-build/src/runtime.rs b/ayllu-build/src/runtime.rs
17index a24b768..6731b2a 100644
18--- a/ayllu-build/src/runtime.rs
19+++ b/ayllu-build/src/runtime.rs
20 @@ -131,7 +131,7 @@ impl Runtime {
21 ) -> Result<(Resolved, i32, BuildGraph), Error> {
22 let resolved = &source.read(&self.config)?;
23 let manifest = &resolved.manifest;
24- db.with(move |mut tx| {
25+ Ok(db.with(move |mut tx| {
26 let manifest_id = tx.manifest_create(
27 manifest,
28 resolved.collection.as_deref(),
29 @@ -145,9 +145,11 @@ impl Runtime {
30 // Allocate components into the database
31 for workflow in manifest.workflows.iter() {
32 if workflow.steps.is_empty() {
33- return Err(Error::EmptyWorkflow {
34- name: workflow.name.clone(),
35- });
36+ return Err(ayllu_database::Error::Rollback(Box::new(
37+ Error::EmptyWorkflow {
38+ name: workflow.name.clone(),
39+ },
40+ )));
41 }
42 let workflow_id = tx.workflow_create(&CreateWorkflowArgs {
43 name: &workflow.name,
44 @@ -189,7 +191,7 @@ impl Runtime {
45 })?;
46 step.environment.iter().try_for_each(|(key, value)| {
47 let _ = tx.variable_create(step_id, key, value)?;
48- Ok::<_, Error>(())
49+ Ok::<_, ayllu_database::Error>(())
50 })?;
51 by_name.insert(step.name.clone(), graph.add_node(Unit::Step(step_id)));
52 let step = tx.step_read(step_id)?;
53 @@ -235,14 +237,16 @@ impl Runtime {
54 }
55
56 if is_cyclic_directed(&graph) {
57- return Err(Error::CycleDetected);
58+ return Err(ayllu_database::Error::Rollback(Box::new(
59+ Error::CycleDetected,
60+ )));
61 }
62
63 let dag_json = serde_json::ser::to_string(&graph).unwrap();
64 tx.manifest_set_dag_content(manifest_id, &dag_json)?;
65
66 Ok((resolved.clone(), manifest_id, graph))
67- })
68+ })?)
69 }
70
71 pub fn shell(&self, manifest_id: i32, workflow_id: i32) -> Result<i32, Error> {
72 diff --git a/ayllu-web/src/error.rs b/ayllu-web/src/error.rs
73index 178df67..55da2ae 100644
74--- a/ayllu-web/src/error.rs
75+++ b/ayllu-web/src/error.rs
76 @@ -66,11 +66,3 @@ impl From<ayllu_database::Error> for Error {
77 }
78 }
79 }
80-
81- impl From<ayllu_database::RawError> for Error {
82- fn from(value: ayllu_database::RawError) -> Self {
83- Error::Database {
84- message: format!("Blob IO Error: {value:?}"),
85- }
86- }
87- }
88 diff --git a/crates/database/src/build.rs b/crates/database/src/build.rs
89index 5bb5b02..008b047 100644
90--- a/crates/database/src/build.rs
91+++ b/crates/database/src/build.rs
92 @@ -144,7 +144,7 @@ pub mod manifests {
93 git_hash: Option<&str>,
94 ) -> Result<i32, Error> {
95 let config_str = serde_json::to_string(config).unwrap();
96- diesel::insert_into(table)
97+ let id = diesel::insert_into(table)
98 .values((
99 dsl::config.eq(config_str),
100 dsl::collection.eq(collection),
101 @@ -153,7 +153,8 @@ pub mod manifests {
102 dsl::state.eq(State::Created),
103 ))
104 .returning(dsl::id)
105- .get_result(self.0)
106+ .get_result(self.0)?;
107+ Ok(id)
108 }
109
110 pub fn manifest_check(&mut self, id: i32) -> Result<Option<State>, Error> {
111 @@ -247,10 +248,11 @@ pub mod manifests {
112 }
113
114 pub fn manifest_read(&mut self, id: i32) -> Result<Manifest, Error> {
115- table
116+ let manifest = table
117 .filter(manifests::dsl::id.eq(id))
118 .select(Manifest::as_select())
119- .get_result(self.0)
120+ .get_result(self.0)?;
121+ Ok(manifest)
122 }
123
124 pub fn manifest_read_config<T: for<'a> Deserialize<'a>>(
125 @@ -286,7 +288,7 @@ pub mod manifests {
126 query = query.filter(dsl::name.eq(name));
127 };
128
129- query.load(self.0)
130+ Ok(query.load(self.0)?)
131 }
132
133 pub fn manifest_latest_id(
134 @@ -387,10 +389,10 @@ pub mod workflows {
135
136 impl Ptr<'_> {
137 pub fn workflow_create(&mut self, args: &CreateWorkflowArgs) -> Result<i32, Error> {
138- diesel::insert_into(table)
139+ Ok(diesel::insert_into(table)
140 .values(args)
141 .returning(dsl::id)
142- .get_result(self.0)
143+ .get_result(self.0)?)
144 }
145
146 pub fn workflow_latest_id(&mut self, manifest_id: i32) -> Result<Option<i32>, Error> {
147 @@ -404,17 +406,17 @@ pub mod workflows {
148 }
149
150 pub fn workflow_read(&mut self, workflow_id: i32) -> Result<Workflow, Error> {
151- table
152+ Ok(table
153 .select(Workflow::as_select())
154 .filter(dsl::id.eq(workflow_id))
155- .first::<Workflow>(self.0)
156+ .first::<Workflow>(self.0)?)
157 }
158
159 pub fn workflow_read_all(&mut self, manifest_id: i32) -> Result<Vec<Workflow>, Error> {
160- table
161+ Ok(table
162 .select(Workflow::as_select())
163 .filter(dsl::manifest_id.eq(manifest_id))
164- .load(self.0)
165+ .load(self.0)?)
166 }
167
168 pub fn workflow_start(&mut self, workflow_id: i32) -> Result<(), Error> {
169 @@ -494,17 +496,17 @@ pub mod steps {
170
171 impl Ptr<'_> {
172 pub fn step_create(&mut self, args: &CreateStepArgs) -> Result<i32, Error> {
173- diesel::insert_into(table)
174+ Ok(diesel::insert_into(table)
175 .values(args)
176 .returning(dsl::id)
177- .get_result(self.0)
178+ .get_result(self.0)?)
179 }
180
181 pub fn step_read(&mut self, id: i32) -> Result<Step, Error> {
182- table
183+ Ok(table
184 .select(Step::as_select())
185 .filter(dsl::id.eq(id))
186- .first::<Step>(self.0)
187+ .first::<Step>(self.0)?)
188 }
189
190 pub fn step_start(&mut self, id: i32) -> Result<(), Error> {
191 @@ -619,17 +621,17 @@ pub mod logs {
192
193 impl Ptr<'_> {
194 pub fn log_write_line(&mut self, args: &WriteLineArgs) -> Result<i32, Error> {
195- diesel::insert_into(table)
196+ Ok(diesel::insert_into(table)
197 .values((args, dsl::timestamp.eq(args.timestamp)))
198 .returning(dsl::id)
199- .get_result(self.0)
200+ .get_result(self.0)?)
201 }
202
203 pub fn log_read(&mut self, step_id: i32, _offset: Option<i32>) -> Result<Vec<Line>, Error> {
204- table
205+ Ok(table
206 .select(Line::as_select())
207 .filter(dsl::step_id.eq(step_id))
208- .load(self.0)
209+ .load(self.0)?)
210 }
211 }
212 }
213 @@ -642,22 +644,18 @@ pub mod outputs {
214
215 use super::*;
216 use crate::{
217- Blob, RawError,
218+ Blob,
219 schema::outputs::{dsl, table},
220 };
221 use diesel::{ExpressionMethods, RunQueryDsl};
222 use rusqlite::MAIN_DB;
223
224 impl Blob {
225- pub fn output_read<W: Write>(
226- &self,
227- output_id: i32,
228- target: &mut W,
229- ) -> Result<u64, RawError> {
230+ pub fn output_read<W: Write>(&self, output_id: i32, target: &mut W) -> Result<u64, Error> {
231 let mut handle =
232 self.0
233 .blob_open(MAIN_DB, "outputs", "content", output_id as i64, false)?;
234- let copied = std::io::copy(&mut handle, target).map_err(|e| RawError::BlobFailed {
235+ let copied = std::io::copy(&mut handle, target).map_err(|e| Error::BlobFailed {
236 error: e,
237 file: None,
238 })?;
239 @@ -669,7 +667,7 @@ pub mod outputs {
240 output_id: i32,
241 input: &mut R,
242 size: u64,
243- ) -> Result<u64, RawError> {
244+ ) -> Result<u64, Error> {
245 // let mut fp = std::fs::OpenOptions::new()
246 // .read(true)
247 // .open(input)
248 @@ -687,7 +685,7 @@ pub mod outputs {
249 let mut handle =
250 self.0
251 .blob_open(MAIN_DB, "outputs", "content", output_id as i64, false)?;
252- let copied = std::io::copy(input, &mut handle).map_err(|e| RawError::BlobFailed {
253+ let copied = std::io::copy(input, &mut handle).map_err(|e| Error::BlobFailed {
254 error: e,
255 file: None,
256 })?;
257 @@ -729,21 +727,21 @@ pub mod outputs {
258 sha256sum: Option<&String>,
259 ) -> Result<i32, Error> {
260 let path = path.to_string_lossy().to_string();
261- diesel::insert_into(table)
262+ Ok(diesel::insert_into(table)
263 .values((
264 dsl::workflow_id.eq(workflow_id),
265 dsl::path.eq(path),
266 dsl::sha256sum.eq(sha256sum),
267 ))
268 .returning(dsl::id)
269- .get_result(self.0)
270+ .get_result(self.0)?)
271 }
272
273 pub fn output_list(&mut self, workflow_id: i32) -> Result<Vec<Output>, Error> {
274- table
275+ Ok(table
276 .select(Output::as_select())
277 .filter(dsl::workflow_id.eq(workflow_id))
278- .load(self.0)
279+ .load(self.0)?)
280 }
281
282 pub fn output_finalize(&mut self, id: i32, digest: &str, size: i32) -> Result<(), Error> {
283 @@ -816,7 +814,7 @@ pub mod cache {
284 } else {
285 name.to_string()
286 };
287- diesel::insert_into(table)
288+ Ok(diesel::insert_into(table)
289 .values((
290 dsl::workflow_id.eq(workflow_id),
291 dsl::name.eq(name),
292 @@ -825,14 +823,14 @@ pub mod cache {
293 dsl::transient.eq(if transient { 1 } else { 0 }),
294 ))
295 .returning(dsl::id)
296- .get_result(self.0)
297+ .get_result(self.0)?)
298 }
299
300 pub fn cache_list(&mut self, workflow_id: i32) -> Result<Vec<Cache>, Error> {
301- table
302+ Ok(table
303 .select(Cache::as_select())
304 .filter(dsl::workflow_id.eq(workflow_id))
305- .load(self.0)
306+ .load(self.0)?)
307 }
308 }
309 }
310 @@ -870,21 +868,21 @@ pub mod variables {
311 name: &str,
312 value: &str,
313 ) -> Result<i32, Error> {
314- diesel::insert_into(table)
315+ Ok(diesel::insert_into(table)
316 .values((
317 dsl::step_id.eq(step_id),
318 dsl::name.eq(name),
319 dsl::value.eq(value),
320 ))
321 .returning(dsl::id)
322- .get_result(self.0)
323+ .get_result(self.0)?)
324 }
325
326 pub fn variable_read_all(&mut self, step_id: i32) -> Result<Vec<Variable>, Error> {
327- table
328+ Ok(table
329 .select(Variable::as_select())
330 .filter(dsl::step_id.eq(step_id))
331- .load(self.0)
332+ .load(self.0)?)
333 }
334 }
335 }
336 @@ -930,7 +928,7 @@ pub mod samples {
337
338 impl Ptr<'_> {
339 pub fn sample_create(&mut self, workflow_id: i32, params: &Params) -> Result<i32, Error> {
340- diesel::insert_into(table)
341+ Ok(diesel::insert_into(table)
342 .values((
343 dsl::workflow_id.eq(workflow_id),
344 dsl::timestamp.eq(params.timestamp),
345 @@ -940,7 +938,7 @@ pub mod samples {
346 dsl::cpu_system_usec.eq(params.cpu_system_usec),
347 ))
348 .returning(dsl::id)
349- .get_result(self.0)
350+ .get_result(self.0)?)
351 }
352
353 pub fn sample_read_all(&mut self, workflow_id: i32) -> Result<Samples, Error> {
354 diff --git a/crates/database/src/error.rs b/crates/database/src/error.rs
355new file mode 100644
356index 0000000..2d906ae
357--- /dev/null
358+++ b/crates/database/src/error.rs
359 @@ -0,0 +1,20 @@
360+ use std::path::PathBuf;
361+
362+ #[derive(Debug, thiserror::Error)]
363+ pub enum Error {
364+ #[error("Raw SQLite Connection Failed: {0}")]
365+ ConnectionRaw(#[from] rusqlite::Error),
366+ #[error("SQLite Connection Failed: {0}")]
367+ Connection(#[from] diesel::ConnectionError),
368+ #[error("Ayllu application error, rolling back: {0}")]
369+ Rollback(Box<dyn std::error::Error + Send + Sync>),
370+ #[error("Blob failed to copy {file:?}: {error}")]
371+ BlobFailed {
372+ file: Option<PathBuf>,
373+ error: std::io::Error,
374+ },
375+ #[error("Diesel query failed: {0}")]
376+ DieselQuery(#[from] diesel::result::Error),
377+ #[error("Migration failed: {0}")]
378+ Migration(String),
379+ }
380 diff --git a/crates/database/src/lib.rs b/crates/database/src/lib.rs
381index 476f4f4..7555212 100644
382--- a/crates/database/src/lib.rs
383+++ b/crates/database/src/lib.rs
384 @@ -1,39 +1,20 @@
385- use std::path::{Path, PathBuf};
386+ use std::path::Path;
387
388 use diesel::{Connection, SqliteConnection};
389 use diesel_migrations::{EmbeddedMigrations, MigrationHarness, embed_migrations};
390
391 pub mod build;
392+ pub mod error;
393 pub mod schema;
394 pub mod tabled_util;
395
396- pub use diesel::result::Error;
397+ pub use crate::error::Error;
398+
399 use rusqlite::OpenFlags;
400
401 const MIGRATIONS: EmbeddedMigrations = embed_migrations!("./migrations");
402
403- #[derive(Debug, thiserror::Error)]
404- pub enum ConnectionError {
405- #[error("Raw SQLite Connection Failed: {0}")]
406- ConnectionRaw(#[from] rusqlite::Error),
407- #[error("SQLite Connection Failed: {0}")]
408- Connection(#[from] diesel::ConnectionError),
409- #[error("Migration failed: {0}")]
410- Migration(String),
411- }
412-
413- #[derive(Debug, thiserror::Error)]
414- pub enum RawError {
415- #[error("Raw SQLite Connection Failed: {0}")]
416- ConnectionRaw(#[from] rusqlite::Error),
417- #[error("Blob failed to copy {file:?}: {error}")]
418- BlobFailed {
419- file: Option<PathBuf>,
420- error: std::io::Error,
421- },
422- }
423-
424- fn connect(connection_url: &Path, read_only: bool) -> Result<SqliteConnection, ConnectionError> {
425+ fn connect(connection_url: &Path, read_only: bool) -> Result<SqliteConnection, Error> {
426 let as_str = format!(
427 "file://{}{}",
428 connection_url.to_string_lossy(),
429 @@ -48,10 +29,7 @@ fn connect(connection_url: &Path, read_only: bool) -> Result<SqliteConnection, C
430 Ok(conn)
431 }
432
433- fn connect_rusqlite(
434- connection_url: &Path,
435- no_lock: bool,
436- ) -> Result<rusqlite::Connection, RawError> {
437+ fn connect_rusqlite(connection_url: &Path, no_lock: bool) -> Result<rusqlite::Connection, Error> {
438 if no_lock {
439 let conn = rusqlite::Connection::open_with_flags(
440 format!("file://{}?nolock=1", connection_url.to_string_lossy()),
441 @@ -65,16 +43,16 @@ fn connect_rusqlite(
442
443 /// Apply all pending migrations.
444 /// TODO: Once Ayllu reaches 1.0 we can make this more flexible.
445- pub fn migrate(connection_url: &Path) -> Result<(), ConnectionError> {
446+ pub fn migrate(connection_url: &Path) -> Result<(), Error> {
447 let mut conn = connect(connection_url, false)?;
448 for migration in conn
449 .pending_migrations(MIGRATIONS)
450- .map_err(|e| ConnectionError::Migration(e.to_string()))?
451+ .map_err(|e| Error::Migration(e.to_string()))?
452 {
453 tracing::info!("Applying migration {}", migration.name());
454 let version = conn
455 .run_migration(&migration)
456- .map_err(|e| ConnectionError::Migration(e.to_string()))?;
457+ .map_err(|e| Error::Migration(e.to_string()))?;
458 tracing::info!("Success: {}", version);
459 }
460 Ok(())
461 @@ -83,7 +61,7 @@ pub fn migrate(connection_url: &Path) -> Result<(), ConnectionError> {
462 pub struct Blob(rusqlite::Connection);
463
464 impl Blob {
465- pub fn new(path: &Path) -> Result<Self, RawError> {
466+ pub fn new(path: &Path) -> Result<Self, Error> {
467 Ok(Self(connect_rusqlite(path, false)?))
468 }
469
470 @@ -91,7 +69,7 @@ impl Blob {
471 /// This should be used in-place of RO mode in the web UI and the UI must
472 /// never perform write operations. The SQLite blob API requires RW access
473 /// even though reading a blob does not require writing.
474- pub fn new_no_lock(path: &Path) -> Result<Self, RawError> {
475+ pub fn new_no_lock(path: &Path) -> Result<Self, Error> {
476 Ok(Self(connect_rusqlite(path, true)?))
477 }
478 }
479 @@ -103,13 +81,13 @@ pub struct Wrapper {
480 }
481
482 impl Wrapper {
483- pub fn new(path: &Path) -> Result<Self, ConnectionError> {
484+ pub fn new(path: &Path) -> Result<Self, Error> {
485 Ok(Self {
486 conn: connect(path, false)?,
487 })
488 }
489
490- pub fn new_ro(path: &Path) -> Result<Self, ConnectionError> {
491+ pub fn new_ro(path: &Path) -> Result<Self, Error> {
492 Ok(Self {
493 conn: connect(path, true)?,
494 })
495 @@ -121,11 +99,20 @@ impl Wrapper {
496 }
497
498 /// Execute the closure in a transaction
499- pub fn with<F, T, E>(&mut self, f: F) -> Result<T, E>
500+ pub fn with<F, T>(&mut self, f: F) -> Result<T, crate::error::Error>
501 where
502- E: From<diesel::result::Error>,
503- F: FnOnce(Ptr) -> Result<T, E>,
504+ F: FnOnce(Ptr) -> Result<T, crate::error::Error>,
505 {
506- self.conn.transaction(move |conn| f(Ptr(conn)))
507+ Ok(self.conn.transaction(move |conn| match f(Ptr(conn)) {
508+ Ok(v) => Ok(v),
509+ Err(e) => match e {
510+ Error::DieselQuery(error) => Err(error),
511+ Error::Rollback(message) => {
512+ tracing::error!("Caller requested rollback: {message}");
513+ Err(diesel::result::Error::RollbackTransaction)
514+ }
515+ _ => unreachable!(),
516+ },
517+ })?)
518 }
519 }
520 diff --git a/quipu/src/error.rs b/quipu/src/error.rs
521index 749fbee..31ea74e 100644
522--- a/quipu/src/error.rs
523+++ b/quipu/src/error.rs
524 @@ -25,8 +25,4 @@ pub enum QuipuError {
525 Git(#[from] ayllu_git::Error),
526 #[error("Database: {0}")]
527 Database(#[from] ayllu_database::Error),
528- #[error("Migration: {0}")]
529- Migration(#[from] ayllu_database::ConnectionError),
530- #[error("Rusqlite: {0}")]
531- Rusqlite(#[from] ayllu_database::RawError),
532 }