git.iko.soy/reshape
on main help sign in
Use advisory lock to avoid conflicts between instances
88da0d2fabianlindfors
src/db.rs+65−3
compare with main
unifiedsplit
@@ -1,5 +1,68 @@
1+use anyhow::anyhow;
1use postgres::{types::ToSql, NoTls, Row};
2use postgres::{types::ToSql, NoTls, Row};
2
3
4+// DbLocker wraps a regular DbConn, only allowing access using the
5+// `lock` method. This method will acquire the advisory lock before
6+// allowing access to the database, and then release it afterwards.
7+//
8+// We use advisory locks to avoid multiple Reshape instances working
9+// on the same database as the same time. DbLocker is the only way to
10+// get a DbConn which ensures that all database access is protected by
11+// a lock.
12+//
13+// Postgres docs on advisory locks:
14+// https://www.postgresql.org/docs/current/explicit-locking.html#ADVISORY-LOCKS
15+pub struct DbLocker {
16+ client: DbConn,
17+}
18+
19+impl DbLocker {
20+ // Advisory lock keys in Postgres are 64-bit integers.
21+ // The key we use was chosen randomly.
22+ const LOCK_KEY: i64 = 4036779288569897133;
23+
24+ pub fn connect(config: &postgres::Config) -> anyhow::Result<Self> {
25+ let pg = config.connect(NoTls)?;
26+ Ok(Self {
27+ client: DbConn::new(pg),
28+ })
29+ }
30+
31+ pub fn lock(
32+ &mut self,
33+ f: impl FnOnce(&mut DbConn) -> anyhow::Result<()>,
34+ ) -> anyhow::Result<()> {
35+ self.acquire_lock()?;
36+ let result = f(&mut self.client);
37+ self.release_lock()?;
38+
39+ result
40+ }
41+
42+ fn acquire_lock(&mut self) -> anyhow::Result<()> {
43+ let success = self
44+ .client
45+ .query(&format!("SELECT pg_try_advisory_lock({})", Self::LOCK_KEY))?
46+ .first()
47+ .ok_or_else(|| anyhow!("unexpectedly failed when acquiring advisory lock"))
48+ .map(|row| row.get::<'_, _, bool>(0))?;
49+
50+ if success {
51+ Ok(())
52+ } else {
53+ Err(anyhow!("another instance of Reshape is already running"))
54+ }
55+ }
56+
57+ fn release_lock(&mut self) -> anyhow::Result<()> {
58+ self.client
59+ .query(&format!("SELECT pg_advisory_unlock({})", Self::LOCK_KEY))?
60+ .first()
61+ .ok_or_else(|| anyhow!("unexpectedly failed when releasing advisory lock"))?;
62+ Ok(())
63+ }
64+}
65+
3pub trait Conn {
66pub trait Conn {
4 fn run(&mut self, query: &str) -> anyhow::Result<()>;
67 fn run(&mut self, query: &str) -> anyhow::Result<()>;
5 fn query(&mut self, query: &str) -> anyhow::Result<Vec<Row>>;
68 fn query(&mut self, query: &str) -> anyhow::Result<Vec<Row>>;
@@ -16,9 +79,8 @@ pub struct DbConn {
16}
79}
17
80
18impl DbConn {
81impl DbConn {
19− pub fn connect(config: &postgres::Config) -> anyhow::Result<DbConn> {
82+ fn new(client: postgres::Client) -> Self {
20− let client = config.connect(NoTls)?;
21− Ok(DbConn { client })
83+ DbConn { client }
22 }
84 }
23}
85}
24
86
src/lib.rs+405−420
@@ -5,7 +5,7 @@ use crate::{
5
5
6use anyhow::{anyhow, Context};
6use anyhow::{anyhow, Context};
7use colored::*;
7use colored::*;
8−use db::{Conn, DbConn};
8+use db::{Conn, DbConn, DbLocker};
9use postgres::Config;
9use postgres::Config;
10use schema::Table;
10use schema::Table;
11
11
@@ -18,8 +18,7 @@ mod state;
18pub use crate::state::State;
18pub use crate::state::State;
19
19
20pub struct Reshape {
20pub struct Reshape {
21− pub state: State,
21+ db: DbLocker,
22− db: DbConn,
23}
22}
24
23
25impl Reshape {
24impl Reshape {
@@ -45,494 +44,480 @@ impl Reshape {
45 }
44 }
46
45
47 fn new_with_config(config: &Config) -> anyhow::Result<Reshape> {
46 fn new_with_config(config: &Config) -> anyhow::Result<Reshape> {
48− let mut db = DbConn::connect(config)?;
47+ let db = DbLocker::connect(config)?;
49− let state = State::load(&mut db)?;
48+ Ok(Reshape { db })
49+ }
50
50
51− Ok(Reshape { db, state })
51+ pub fn migrate(
52+ &mut self,
53+ migrations: impl IntoIterator<Item = Migration>,
54+ ) -> anyhow::Result<()> {
55+ self.db.lock(|db| {
56+ let mut state = State::load(db)?;
57+ migrate(db, &mut state, migrations)
58+ })
52 }
59 }
53
60
54− pub fn migrate<T>(&mut self, migrations: T) -> anyhow::Result<()>
61+ pub fn complete(&mut self) -> anyhow::Result<()> {
55− where
62+ self.db.lock(|db| {
56− T: IntoIterator<Item = Migration>,
57− {
58− self.state = State::load(&mut self.db)?;
63+ let mut state = State::load(db)?;
64+ complete(db, &mut state)
65+ })
66+ }
59
67
60− // Make sure no migration is in progress
68+ pub fn abort(&mut self) -> anyhow::Result<()> {
61− if let State::InProgress { .. } = &self.state {
69+ self.db.lock(|db| {
62− println!("Migration already in progress, please complete using 'reshape complete'");
70+ let mut state = State::load(db)?;
63− return Ok(());
71+ abort(db, &mut state)
72+ })
64− }
73+ }
65
74
66− if let State::Completing { .. } = &self.state {
75+ pub fn remove(&mut self) -> anyhow::Result<()> {
67− println!(
76+ self.db.lock(|db| {
68− "Migration already in progress and has started completion, please finish using 'reshape complete'"
77+ let mut state = State::load(db)?;
69− );
70− return Ok(());
71− }
72
78
73− // Determine which migrations need to be applied by comparing the provided migrations
79+ // Remove migration schemas and views
74− // with the already applied ones stored in the state. This will throw an error if the
75− // two sets of migrations don't agree, for example if a new migration has been added
76− // in between two existing ones.
77− let current_migration = state::current_migration(&mut self.db)?;
80+ if let Some(current_migration) = &state::current_migration(db)? {
78− let remaining_migrations = state::remaining_migrations(&mut self.db, migrations)?;
81+ db.run(&format!(
79− if remaining_migrations.is_empty() {
82+ "DROP SCHEMA IF EXISTS {} CASCADE",
80− println!("No migrations left to apply");
83+ schema_name_for_migration(current_migration)
81− return Ok(());
84+ ))?;
82− }
85+ }
83
86
84− // If we have already started applying some migrations we need to ensure that
87+ if let State::InProgress { migrations } = &state {
85− // they are the same ones we want to apply now
88+ let target_migration = migrations.last().unwrap().name.to_string();
86− if let State::Applying {
89+ db.run(&format!(
87− migrations: existing_migrations,
90+ "DROP SCHEMA IF EXISTS {} CASCADE",
88− } = &self.state
91+ schema_name_for_migration(&target_migration)
89− {
90− if existing_migrations != &remaining_migrations {
91− return Err(anyhow!(
92− "a previous migration seems to have failed without cleaning up. Please run `reshape abort` and then run migrate again."
93− ));
92+ ))?;
94 }
93 }
95− }
96
94
97− // Move to the "Applying" state which is necessary as we can't run the migrations
95+ // Remove all tables
98− // and state update as a single transaction. If a migration unexpectedly fails without
96+ let schema = Schema::new();
99− // automatically aborting, this state saves us from dangling migrations. It forces the user
97+ for table in schema.get_tables(db)? {
100− // to either run migrate again (which works as all migrations are idempotent) or abort.
98+ db.run(&format!(
101− self.state.applying(remaining_migrations.clone());
99+ r#"
102− self.state.save(&mut self.db)?;
100+ DROP TABLE IF EXISTS "{}" CASCADE
101+ "#,
102+ table.real_name
103+ ))?;
104+ }
103
105
104− println!("Applying {} migrations\n", remaining_migrations.len());
106+ // Reset state
107+ state.clear(db)?;
105
108
106− helpers::set_up_helpers(&mut self.db, &current_migration)
109+ println!("Reshape and all data has been removed");
107− .context("failed to set up helpers")?;
108
110
109− let mut new_schema = Schema::new();
111+ Ok(())
110− let mut last_migration_index = usize::MAX;
112+ })
111− let mut last_action_index = usize::MAX;
113+ }
112− let mut result: anyhow::Result<()> = Ok(());
114+}
115+
116+pub fn latest_schema_from_migrations(migrations: &[Migration]) -> Option<String> {
117+ migrations
118+ .last()
119+ .map(|migration| schema_name_for_migration(&migration.name))
120+}
113
121
114− for (migration_index, migration) in remaining_migrations.iter().enumerate() {
122+pub fn schema_query_for_migration(migration_name: &str) -> String {
115− println!("Migrating '{}':", migration.name);
123+ let schema_name = schema_name_for_migration(migration_name);
116− last_migration_index = migration_index;
124+ format!("SET search_path TO {}", schema_name)
125+}
117
126
118− for (action_index, action) in migration.actions.iter().enumerate() {
127+fn schema_name_for_migration(migration_name: &str) -> String {
119− last_action_index = action_index;
128+ format!("migration_{}", migration_name)
129+}
120
130
121− let description = action.describe();
131+fn migrate(
122− print!(" + {} ", description);
132+ db: &mut DbConn,
133+ state: &mut State,
134+ migrations: impl IntoIterator<Item = Migration>,
135+) -> anyhow::Result<()> {
136+ // Make sure no migration is in progress
137+ if let State::InProgress { .. } = &state {
138+ println!("Migration already in progress, please complete using 'reshape complete'");
139+ return Ok(());
140+ }
123
141
124− let ctx = MigrationContext::new(migration_index, action_index);
142+ if let State::Completing { .. } = &state {
125− result = action
143+ println!(
126− .run(&ctx, &mut self.db, &new_schema)
144+ "Migration already in progress and has started completion, please finish using 'reshape complete'"
127− .with_context(|| format!("failed to {}", description));
145+ );
146+ return Ok(());
147+ }
128
148
129− if result.is_ok() {
149+ // Determine which migrations need to be applied by comparing the provided migrations
130− action.update_schema(&ctx, &mut new_schema);
150+ // with the already applied ones stored in the state. This will throw an error if the
131− println!("{}", "done".green());
151+ // two sets of migrations don't agree, for example if a new migration has been added
132− } else {
152+ // in between two existing ones.
133− println!("{}", "failed".red());
153+ let current_migration = state::current_migration(db)?;
134− break;
154+ let remaining_migrations = state::remaining_migrations(db, migrations)?;
135− }
155+ if remaining_migrations.is_empty() {
156+ println!("No migrations left to apply");
157+ return Ok(());
136− }
158+ }
137
159
138− println!();
160+ // If we have already started applying some migrations we need to ensure that
161+ // they are the same ones we want to apply now
162+ if let State::Applying {
163+ migrations: existing_migrations,
164+ } = &state
165+ {
166+ if existing_migrations != &remaining_migrations {
167+ return Err(anyhow!(
168+ "a previous migration seems to have failed without cleaning up. Please run `reshape abort` and then run migrate again."
169+ ));
139 }
170 }
171+ }
140
172
141− // If a migration failed, we abort all the migrations that were applied
173+ // Move to the "Applying" state which is necessary as we can't run the migrations
142− if let Err(err) = result {
174+ // and state update as a single transaction. If a migration unexpectedly fails without
143− println!("A migration failed, aborting migrations that have already been applied");
175+ // automatically aborting, this state saves us from dangling migrations. It forces the user
176+ // to either run migrate again (which works as all migrations are idempotent) or abort.
177+ state.applying(remaining_migrations.clone());
178+ state.save(db)?;
144
179
145− // Set to the Aborting state. This is to ensure that the failed
146− // migration is fully aborted and nothing is left dangling.
147− // If the abort is interrupted for any reason, the user can try again
148− // by running `reshape abort`.
149− self.state.aborting(
150− remaining_migrations.clone(),
180+ println!("Applying {} migrations\n", remaining_migrations.len());
151− last_migration_index + 1,
152− last_action_index + 1,
153− );
154
181
155− // Abort will only
182+ helpers::set_up_helpers(db, &current_migration).context("failed to set up helpers")?;
156− self.abort()?;
157
183
158− return Err(err);
184+ let mut new_schema = Schema::new();
159− }
185+ let mut last_migration_index = usize::MAX;
186+ let mut last_action_index = usize::MAX;
187+ let mut result: anyhow::Result<()> = Ok(());
160
188
161− // Create schema and views for migration
189+ for (migration_index, migration) in remaining_migrations.iter().enumerate() {
162− let target_migration = remaining_migrations.last().unwrap().name.to_string();
190+ println!("Migrating '{}':", migration.name);
163− self.create_schema_for_migration(&target_migration, &new_schema)
191+ last_migration_index = migration_index;
164− .with_context(|| {
165− format!("failed to create schema for migration {}", target_migration)
166− })?;
167−
168− // Update state once migrations have been performed
169− self.state.in_progress(remaining_migrations);
170− self.state
171− .save(&mut self.db)
172− .context("failed to save in-progress state")?;
173−
174− // If we started from a blank slate, we can finish the migration immediately
175− if current_migration.is_none() {
176− println!("Automatically completing migrations\n");
177− self.complete_migration()
178− .context("failed to automatically complete migrations")?;
179−
180− println!("Migrations complete:");
181− println!(
182− " - Run '{}' from your application to use the latest schema",
183− schema_query_for_migration(&target_migration)
184− );
185− } else {
186− println!("Migrations have been applied and the new schema is ready for use:");
187− println!(
188− " - Run '{}' from your application to use the latest schema",
189− schema_query_for_migration(&target_migration)
190− );
191− println!(
192− " - Run 'reshape complete' once your application has been updated and the previous schema is no longer in use"
193− );
194− }
195
192
196− Ok(())
193+ for (action_index, action) in migration.actions.iter().enumerate() {
197− }
194+ last_action_index = action_index;
198
195
199− pub fn complete_migration(&mut self) -> anyhow::Result<()> {
196+ let description = action.describe();
200− // Make sure a migration is in progress
197+ print!(" + {} ", description);
201− let (remaining_migrations, starting_migration_index, starting_action_index) = match self.state.clone() {
202− State::InProgress { migrations } => {
203− // Move into the Completing state. Once in this state,
204− // the migration can't be aborted and must be completed.
205− self.state.completing(migrations.clone(), 0, 0);
206− self.state.save(&mut self.db).context("failed to save state")?;
207−
208− (migrations, 0, 0)
209− },
210− State::Completing {
211− migrations,
212− current_migration_index,
213− current_action_index
214− } => (migrations, current_migration_index, current_action_index),
215− State::Aborting { .. } => {
216− return Err(anyhow!("migration been aborted and can't be completed. Please finish using `reshape abort`."))
217− }
218− State::Applying { .. } => {
219− return Err(anyhow!("a previous migration unexpectedly failed. Please run `reshape migrate` to try applying the migration again."))
220− }
221− State::Idle => {
222− println!("No migration in progress");
223− return Ok(());
224− }
225− };
226
198
227− // Remove previous migration's schema
199+ let ctx = MigrationContext::new(migration_index, action_index);
228− if let Some(current_migration) = &state::current_migration(&mut self.db)? {
200+ result = action
229− self.db
201+ .run(&ctx, db, &new_schema)
230− .run(&format!(
202+ .with_context(|| format!("failed to {}", description));
231− "DROP SCHEMA IF EXISTS {} CASCADE",
232− schema_name_for_migration(current_migration)
233− ))
234− .context("failed to remove previous migration's schema")?;
235− }
236
203
237− for (migration_index, migration) in remaining_migrations.iter().enumerate() {
204+ if result.is_ok() {
238− // Skip all the migrations which have already been completed
205+ action.update_schema(&ctx, &mut new_schema);
239− if migration_index < starting_migration_index {
206+ println!("{}", "done".green());
240− continue;
207+ } else {
208+ println!("{}", "failed".red());
209+ break;
241 }
210 }
211+ }
242
212
243− println!("Completing '{}':", migration.name);
213+ println!();
214+ }
244
215
245− for (action_index, action) in migration.actions.iter().enumerate() {
216+ // If a migration failed, we abort all the migrations that were applied
246− // Skip all actions which have already been completed
217+ if let Err(err) = result {
247− if migration_index == starting_migration_index
218+ println!("A migration failed, aborting migrations that have already been applied");
248− && action_index < starting_action_index
249− {
250− continue;
251− }
252
219
253− let description = action.describe();
220+ // Set to the Aborting state. This is to ensure that the failed
254− print!(" + {} ", description);
221+ // migration is fully aborted and nothing is left dangling.
222+ // If the abort is interrupted for any reason, the user can try again
223+ // by running `reshape abort`.
224+ state.aborting(
225+ remaining_migrations.clone(),
226+ last_migration_index + 1,
227+ last_action_index + 1,
228+ );
255
229
256− let ctx = MigrationContext::new(migration_index, action_index);
230+ // Abort will only
231+ abort(db, state)?;
257
232
258− // Update state to indicate that this action has been completed.
233+ return Err(err);
259− // We won't save this new state until after the action has completed.
234+ }
260− self.state.completing(
235+
261− remaining_migrations.clone(),
236+ // Create schema and views for migration
262− migration_index + 1,
237+ let target_migration = remaining_migrations.last().unwrap().name.to_string();
263− action_index + 1,
238+ create_schema_for_migration(db, &target_migration, &new_schema)
239+ .with_context(|| format!("failed to create schema for migration {}", target_migration))?;
240+
241+ // Update state once migrations have been performed
242+ state.in_progress(remaining_migrations);
243+ state.save(db).context("failed to save in-progress state")?;
244+
245+ // If we started from a blank slate, we can finish the migration immediately
246+ if current_migration.is_none() {
247+ println!("Automatically completing migrations\n");
248+ complete(db, state).context("failed to automatically complete migrations")?;
249+
250+ println!("Migrations complete:");
251+ println!(
252+ " - Run '{}' from your application to use the latest schema",
253+ schema_query_for_migration(&target_migration)
254+ );
255+ } else {
256+ println!("Migrations have been applied and the new schema is ready for use:");
257+ println!(
258+ " - Run '{}' from your application to use the latest schema",
259+ schema_query_for_migration(&target_migration)
260+ );
261+ println!(
262+ " - Run 'reshape complete' once your application has been updated and the previous schema is no longer in use"
264 );
263 );
264+ }
265
265
266− // This did_save check is necessary because of the borrow checker.
266+ Ok(())
267− // The Transaction which might be returned from action.complete
268− // contains a mutable reference to self.db. We need the Transaction
269− // to be dropped before we can save the state using self.db instead,
270− // which we achieve here by limiting the lifetime of the Transaction
271− // with a new block.
272− let did_save = {
273− let result = action
274− .complete(&ctx, &mut self.db)
275− .with_context(|| format!("failed to complete migration {}", migration.name))
276− .with_context(|| format!("failed to complete action: {}", description));
277−
278− let maybe_transaction = match result {
279− Ok(maybe_transaction) => {
280− println!("{}", "done".green());
281− maybe_transaction
282− }
267+}
283− Err(e) => {
284− println!("{}", "failed".red());
285− return Err(e);
286− }
287− };
288−
289− // Update state with which migrations and actions have been completed.
290− // Each action can create and return a transaction if they need atomicity.
291− // We use this transaction to update the state to ensure the action only completes.
292− // once.
293− // We want to use a single transaction for each action to keep the length of
294− // the transaction as short as possible. Wherever possible, we don't want to
295− // use a transaction at all.
296− if let Some(mut transaction) = maybe_transaction {
297− self.state
298− .save(&mut transaction)
299− .context("failed to save state after completing action")?;
300− transaction
301− .commit()
302− .context("failed to commit transaction")?;
303−
304− true
305− } else {
306− false
307− }
308− };
309
268
310− // If the action didn't return a transaction we save the state normally instead
269+fn complete(db: &mut DbConn, state: &mut State) -> anyhow::Result<()> {
311− if !did_save {
270+ // Make sure a migration is in progress
312− self.state
271+ let (remaining_migrations, starting_migration_index, starting_action_index) = match state.clone() {
313− .save(&mut self.db)
272+ State::InProgress { migrations } => {
273+ // Move into the Completing state. Once in this state,
274+ // the migration can't be aborted and must be completed.
275+ state.completing(migrations.clone(), 0, 0);
314− .context("failed to save state after completing action")?;
276+ state.save(db).context("failed to save state")?;
277+
278+ (migrations, 0, 0)
279+ },
280+ State::Completing {
281+ migrations,
282+ current_migration_index,
283+ current_action_index
284+ } => (migrations, current_migration_index, current_action_index),
285+ State::Aborting { .. } => {
286+ return Err(anyhow!("migration been aborted and can't be completed. Please finish using `reshape abort`."))
315 }
287 }
288+ State::Applying { .. } => {
289+ return Err(anyhow!("a previous migration unexpectedly failed. Please run `reshape migrate` to try applying the migration again."))
290+ }
291+ State::Idle => {
292+ println!("No migration in progress");
293+ return Ok(());
294+ }
316− }
295+ };
317
296
318− println!();
297+ // Remove previous migration's schema
298+ if let Some(current_migration) = &state::current_migration(db)? {
299+ db.run(&format!(
300+ "DROP SCHEMA IF EXISTS {} CASCADE",
301+ schema_name_for_migration(current_migration)
302+ ))
303+ .context("failed to remove previous migration's schema")?;
304+ }
305+
306+ for (migration_index, migration) in remaining_migrations.iter().enumerate() {
307+ // Skip all the migrations which have already been completed
308+ if migration_index < starting_migration_index {
309+ continue;
319 }
310 }
320
311
321− // Remove helpers which are no longer in use
312+ println!("Completing '{}':", migration.name);
322− helpers::tear_down_helpers(&mut self.db).context("failed to tear down helpers")?;
323
313
324− self.state
314+ for (action_index, action) in migration.actions.iter().enumerate() {
325− .complete(&mut self.db)
315+ // Skip all actions which have already been completed
326− .context("failed to update state as completed")?;
316+ if migration_index == starting_migration_index && action_index < starting_action_index {
317+ continue;
318+ }
327
319
328− Ok(())
320+ let description = action.describe();
329− }
321+ print!(" + {} ", description);
330
322
331− fn create_schema_for_migration(
323+ let ctx = MigrationContext::new(migration_index, action_index);
332− &mut self,
333− migration_name: &str,
334− schema: &Schema,
335− ) -> anyhow::Result<()> {
336− // Create schema for migration
337− let schema_name = schema_name_for_migration(migration_name);
338− self.db
339− .run(&format!("CREATE SCHEMA IF NOT EXISTS {}", schema_name))
340− .with_context(|| {
341− format!(
342− "failed to create schema {} for migration {}",
343− schema_name, migration_name
344− )
345− })?;
346−
347− // Create views inside schema
348− for table in schema.get_tables(&mut self.db)? {
349− Self::create_view_for_table(&mut self.db, &table, &schema_name)?;
350− }
351
324
352− Ok(())
325+ // Update state to indicate that this action has been completed.
353− }
326+ // We won't save this new state until after the action has completed.
327+ state.completing(
328+ remaining_migrations.clone(),
329+ migration_index + 1,
330+ action_index + 1,
331+ );
354
332
355− fn create_view_for_table(
333+ // This did_save check is necessary because of the borrow checker.
356− db: &mut impl Conn,
334+ // The Transaction which might be returned from action.complete
357− table: &Table,
335+ // contains a mutable reference to self.db. We need the Transaction
358− schema: &str,
336+ // to be dropped before we can save the state using self.db instead,
359− ) -> anyhow::Result<()> {
337+ // which we achieve here by limiting the lifetime of the Transaction
360− let select_columns: Vec<String> = table
338+ // with a new block.
361− .columns
339+ let did_save = {
362− .iter()
340+ let result = action
363− .map(|column| {
341+ .complete(&ctx, db)
364− format!(
342+ .with_context(|| format!("failed to complete migration {}", migration.name))
365− r#"
343+ .with_context(|| format!("failed to complete action: {}", description));
366− "{real_name}" AS "{alias}"
344+
367− "#,
345+ let maybe_transaction = match result {
368− real_name = column.real_name,
346+ Ok(maybe_transaction) => {
369− alias = column.name,
347+ println!("{}", "done".green());
370− )
348+ maybe_transaction
371− })
349+ }
372− .collect();
350+ Err(e) => {
351+ println!("{}", "failed".red());
352+ return Err(e);
353+ }
354+ };
373
355
374− db.run(&format!(
356+ // Update state with which migrations and actions have been completed.
375− r#"
357+ // Each action can create and return a transaction if they need atomicity.
376− CREATE OR REPLACE VIEW {schema}."{view_name}" AS
358+ // We use this transaction to update the state to ensure the action only completes.
377− SELECT {columns}
359+ // once.
378− FROM "{table_name}"
360+ // We want to use a single transaction for each action to keep the length of
379− "#,
361+ // the transaction as short as possible. Wherever possible, we don't want to
380− schema = schema,
362+ // use a transaction at all.
381− table_name = table.real_name,
363+ if let Some(mut transaction) = maybe_transaction {
382− view_name = table.name,
364+ state
383− columns = select_columns.join(","),
365+ .save(&mut transaction)
384− ))
366+ .context("failed to save state after completing action")?;
385− .with_context(|| format!("failed to create view for table {}", table.name))?;
367+ transaction
368+ .commit()
369+ .context("failed to commit transaction")?;
386
370
387− Ok(())
371+ true
372+ } else {
373+ false
374+ }
388− }
375+ };
389
376
390− pub fn remove(&mut self) -> anyhow::Result<()> {
377+ // If the action didn't return a transaction we save the state normally instead
391− // Remove migration schemas and views
378+ if !did_save {
392− if let Some(current_migration) = &state::current_migration(&mut self.db)? {
379+ state
393− self.db.run(&format!(
380+ .save(db)
394− "DROP SCHEMA IF EXISTS {} CASCADE",
381+ .context("failed to save state after completing action")?;
395− schema_name_for_migration(current_migration)
382+ }
396− ))?;
397 }
383 }
398
384
399− if let State::InProgress { migrations } = &self.state {
385+ println!();
400− let target_migration = migrations.last().unwrap().name.to_string();
401− self.db.run(&format!(
402− "DROP SCHEMA IF EXISTS {} CASCADE",
403− schema_name_for_migration(&target_migration)
404− ))?;
405− }
386+ }
406
387
407− // Remove all tables
388+ // Remove helpers which are no longer in use
408− let schema = Schema::new();
389+ helpers::tear_down_helpers(db).context("failed to tear down helpers")?;
409− for table in schema.get_tables(&mut self.db)? {
410− self.db.run(&format!(
411− r#"
412− DROP TABLE IF EXISTS "{}" CASCADE
413− "#,
414− table.real_name
415− ))?;
416− }
417
390
418− // Reset state
391+ state
419− self.state.clear(&mut self.db)?;
392+ .complete(db)
393+ .context("failed to update state as completed")?;
420
394
421− println!("Reshape and all data has been removed");
395+ Ok(())
396+}
422
397
423− Ok(())
398+fn abort(db: &mut DbConn, state: &mut State) -> anyhow::Result<()> {
424− }
399+ let (remaining_migrations, last_migration_index, last_action_index) = match state.clone() {
400+ State::InProgress { migrations } | State::Applying { migrations } => {
401+ // Set to the Aborting state. Once this is done, the migration has to
402+ // be fully aborted and can't be completed.
403+ state.aborting(migrations.clone(), 0, 0);
404+ state.save(db)?;
425
405
426− pub fn abort(&mut self) -> anyhow::Result<()> {
427− let (remaining_migrations, last_migration_index, last_action_index) = match self
428− .state
429− .clone()
430− {
431− State::InProgress { migrations } | State::Applying { migrations } => {
432− // Set to the Aborting state. Once this is done, the migration has to
433− // be fully aborted and can't be completed.
434− self.state.aborting(migrations.clone(), 0, 0);
435− self.state.save(&mut self.db)?;
436−
437− (migrations, usize::MAX, usize::MAX)
406+ (migrations, usize::MAX, usize::MAX)
438− }
407+ }
439− State::Aborting {
408+ State::Aborting {
440− migrations,
441− last_migration_index,
442− last_action_index,
443− } => (migrations, last_migration_index, last_action_index),
444− State::Completing { .. } => {
445− return Err(anyhow!("Migration completion has already been started. Please run `reshape complete` again to finish it."));
446− }
447− State::Idle => {
448− println!("No migration is in progress");
449− return Ok(());
450− }
451− };
452−
453− // Remove new migration's schema
454− let target_migration = remaining_migrations.last().unwrap().name.to_string();
455− let schema_name = schema_name_for_migration(&target_migration);
456− self.db
457− .run(&format!("DROP SCHEMA IF EXISTS {} CASCADE", schema_name,))
458− .with_context(|| format!("failed to drop schema {}", schema_name))?;
459−
460− // Abort all pending migrations
409+ migrations,
461− self.abort_migrations(
462− &remaining_migrations,
463 last_migration_index,
410 last_migration_index,
464 last_action_index,
411 last_action_index,
465− )?;
412+ } => (migrations, last_migration_index, last_action_index),
466−
413+ State::Completing { .. } => {
467− helpers::tear_down_helpers(&mut self.db).context("failed to tear down helpers")?;
414+ return Err(anyhow!("Migration completion has already been started. Please run `reshape complete` again to finish it."));
468−
415+ }
469− self.state = State::Idle;
416+ State::Idle => {
470− self.state
417+ println!("No migration is in progress");
471− .save(&mut self.db)
418+ return Ok(());
472− .context("failed to save state")?;
419+ }
420+ };
421+
422+ // Remove new migration's schema
423+ let target_migration = remaining_migrations.last().unwrap().name.to_string();
424+ let schema_name = schema_name_for_migration(&target_migration);
425+ db.run(&format!("DROP SCHEMA IF EXISTS {} CASCADE", schema_name,))
426+ .with_context(|| format!("failed to drop schema {}", schema_name))?;
427+
428+ // Abort all pending migrations
429+ // Abort all migrations in reverse order
430+ for (migration_index, migration) in remaining_migrations.iter().enumerate().rev() {
431+ // Skip migrations which shouldn't be aborted
432+ // The reason can be that they have already been aborted or that
433+ // the migration was never applied in the first place.
434+ if migration_index >= last_migration_index {
435+ continue;
436+ }
473
437
474− Ok(())
438+ print!("Aborting '{}' ", migration.name);
475− }
476
439
477− fn abort_migrations(
478− &mut self,
479− migrations: &[Migration],
480− upper_migration_index: usize,
481− upper_action_index: usize,
482− ) -> anyhow::Result<()> {
483− // Abort all migrations in reverse order
484− for (migration_index, migration) in migrations.iter().enumerate().rev() {
440+ for (action_index, action) in migration.actions.iter().enumerate().rev() {
485− // Skip migrations which shouldn't be aborted
441+ // Skip actions which shouldn't be aborted
486 // The reason can be that they have already been aborted or that
442 // The reason can be that they have already been aborted or that
487− // the migration was never applied in the first place.
443+ // the action was never applied in the first place.
488− if migration_index >= upper_migration_index {
444+ if migration_index == last_migration_index - 1 && action_index >= last_action_index {
489 continue;
445 continue;
490 }
446 }
491
447
492− print!("Aborting '{}' ", migration.name);
493−
494− for (action_index, action) in migration.actions.iter().enumerate().rev() {
495− // Skip actions which shouldn't be aborted
496− // The reason can be that they have already been aborted or that
497− // the action was never applied in the first place.
498− if migration_index == upper_migration_index - 1
499− && action_index >= upper_action_index
500− {
501− continue;
502− }
503−
504− let ctx = MigrationContext::new(migration_index, action_index);
448+ let ctx = MigrationContext::new(migration_index, action_index);
505− action
449+ action
506− .abort(&ctx, &mut self.db)
450+ .abort(&ctx, db)
507− .with_context(|| format!("failed to abort migration {}", migration.name))
451+ .with_context(|| format!("failed to abort migration {}", migration.name))
508− .with_context(|| format!("failed to abort action: {}", action.describe()))?;
452+ .with_context(|| format!("failed to abort action: {}", action.describe()))?;
509−
510− // Update state with which migrations and actions have been aborted.
511− // We don't need to run this in a transaction as aborts are idempotent.
512− self.state
513− .aborting(migrations.to_vec(), migration_index, action_index);
514− self.state
515− .save(&mut self.db)
516− .context("failed to save state")?;
517− }
518
453
519− println!("{}", "done".green());
454+ // Update state with which migrations and actions have been aborted.
455+ // We don't need to run this in a transaction as aborts are idempotent.
456+ state.aborting(remaining_migrations.to_vec(), migration_index, action_index);
457+ state.save(db).context("failed to save state")?;
520 }
458 }
521− Ok(())
459+
460+ println!("{}", "done".green());
522 }
461 }
523−}
524
462
525−pub fn latest_schema_from_migrations(migrations: &[Migration]) -> Option<String> {
463+ helpers::tear_down_helpers(db).context("failed to tear down helpers")?;
526− migrations
464+
527− .last()
465+ *state = State::Idle;
528− .map(|migration| schema_name_for_migration(&migration.name))
466+ state.save(db).context("failed to save state")?;
467+
468+ Ok(())
529}
469}
530
470
471+fn create_schema_for_migration(
472+ db: &mut DbConn,
531−pub fn schema_query_for_migration(migration_name: &str) -> String {
473+ migration_name: &str,
474+ schema: &Schema,
475+) -> anyhow::Result<()> {
476+ // Create schema for migration
532 let schema_name = schema_name_for_migration(migration_name);
477 let schema_name = schema_name_for_migration(migration_name);
533− format!("SET search_path TO {}", schema_name)
478+ db.run(&format!("CREATE SCHEMA IF NOT EXISTS {}", schema_name))
479+ .with_context(|| {
480+ format!(
481+ "failed to create schema {} for migration {}",
482+ schema_name, migration_name
483+ )
484+ })?;
485+
486+ // Create views inside schema
487+ for table in schema.get_tables(db)? {
488+ create_view_for_table(db, &table, &schema_name)?;
489+ }
490+
491+ Ok(())
534}
492}
535
493
536−fn schema_name_for_migration(migration_name: &str) -> String {
494+fn create_view_for_table(db: &mut impl Conn, table: &Table, schema: &str) -> anyhow::Result<()> {
537− format!("migration_{}", migration_name)
495+ let select_columns: Vec<String> = table
496+ .columns
497+ .iter()
498+ .map(|column| {
499+ format!(
500+ r#"
501+ "{real_name}" AS "{alias}"
502+ "#,
503+ real_name = column.real_name,
504+ alias = column.name,
505+ )
506+ })
507+ .collect();
508+
509+ db.run(&format!(
510+ r#"
511+ CREATE OR REPLACE VIEW {schema}."{view_name}" AS
512+ SELECT {columns}
513+ FROM "{table_name}"
514+ "#,
515+ schema = schema,
516+ table_name = table.real_name,
517+ view_name = table.name,
518+ columns = select_columns.join(","),
519+ ))
520+ .with_context(|| format!("failed to create view for table {}", table.name))?;
521+
522+ Ok(())
538}
523}
src/main.rs+2−2
@@ -72,14 +72,14 @@ fn run(opts: Opts) -> anyhow::Result<()> {
72
72
73 // Automatically complete migration if --complete flag is set
73 // Automatically complete migration if --complete flag is set
74 if opts.complete {
74 if opts.complete {
75− reshape.complete_migration()?;
75+ reshape.complete()?;
76 }
76 }
77
77
78 Ok(())
78 Ok(())
79 }
79 }
80 Command::Complete(opts) => {
80 Command::Complete(opts) => {
81 let mut reshape = reshape_from_connection_options(&opts)?;
81 let mut reshape = reshape_from_connection_options(&opts)?;
82− reshape.complete_migration()
82+ reshape.complete()
83 }
83 }
84 Command::Remove(opts) => {
84 Command::Remove(opts) => {
85 let mut reshape = reshape_from_connection_options(&opts)?;
85 let mut reshape = reshape_from_connection_options(&opts)?;
tests/add_column.rs+3−3
@@ -112,7 +112,7 @@ fn add_column() {
112 (first_name.as_ref(), last_name.as_ref())
112 (first_name.as_ref(), last_name.as_ref())
113 );
113 );
114
114
115− reshape.complete_migration().unwrap();
115+ reshape.complete().unwrap();
116 common::assert_cleaned_up(&mut new_db);
116 common::assert_cleaned_up(&mut new_db);
117}
117}
118
118
@@ -205,7 +205,7 @@ fn add_column_nullable() {
205 .simple_query("INSERT INTO users (id, name) VALUES (4, 'Test Testsson'), (5, NULL)")
205 .simple_query("INSERT INTO users (id, name) VALUES (4, 'Test Testsson'), (5, NULL)")
206 .unwrap();
206 .unwrap();
207
207
208− reshape.complete_migration().unwrap();
208+ reshape.complete().unwrap();
209 common::assert_cleaned_up(&mut new_db);
209 common::assert_cleaned_up(&mut new_db);
210}
210}
211
211
@@ -288,6 +288,6 @@ fn add_column_with_default() {
288 .unwrap();
288 .unwrap();
289 assert_eq!("DEFAULT", name);
289 assert_eq!("DEFAULT", name);
290
290
291− reshape.complete_migration().unwrap();
291+ reshape.complete().unwrap();
292 common::assert_cleaned_up(&mut new_db);
292 common::assert_cleaned_up(&mut new_db);
293}
293}
tests/add_index.rs+1−1
@@ -64,6 +64,6 @@ fn add_index() {
64
64
65 assert_eq!(vec![(true, true)], result);
65 assert_eq!(vec![(true, true)], result);
66
66
67− reshape.complete_migration().unwrap();
67+ reshape.complete().unwrap();
68 common::assert_cleaned_up(&mut new_db);
68 common::assert_cleaned_up(&mut new_db);
69}
69}
tests/alter_column.rs+6−6
@@ -99,7 +99,7 @@ fn alter_column_data() {
99 .unwrap();
99 .unwrap();
100 assert_eq!("test testsson", result.get::<_, &str>("name"));
100 assert_eq!("test testsson", result.get::<_, &str>("name"));
101
101
102− reshape.complete_migration().unwrap();
102+ reshape.complete().unwrap();
103 common::assert_cleaned_up(&mut new_db);
103 common::assert_cleaned_up(&mut new_db);
104}
104}
105
105
@@ -198,7 +198,7 @@ fn alter_column_set_not_null() {
198 .unwrap();
198 .unwrap();
199 assert_eq!("Jane Doe", result.get::<_, &str>("name"));
199 assert_eq!("Jane Doe", result.get::<_, &str>("name"));
200
200
201− reshape.complete_migration().unwrap();
201+ reshape.complete().unwrap();
202 common::assert_cleaned_up(&mut new_db);
202 common::assert_cleaned_up(&mut new_db);
203}
203}
204
204
@@ -280,7 +280,7 @@ fn alter_column_rename() {
280 .map(|row| row.get::<_, String>("full_name"))
280 .map(|row| row.get::<_, String>("full_name"))
281 .eq(expected));
281 .eq(expected));
282
282
283− reshape.complete_migration().unwrap();
283+ reshape.complete().unwrap();
284 common::assert_cleaned_up(&mut new_db);
284 common::assert_cleaned_up(&mut new_db);
285}
285}
286
286
@@ -402,7 +402,7 @@ fn alter_column_multiple() {
402 .unwrap();
402 .unwrap();
403 assert_eq!(48, result);
403 assert_eq!(48, result);
404
404
405− reshape.complete_migration().unwrap();
405+ reshape.complete().unwrap();
406 common::assert_cleaned_up(&mut new_db);
406 common::assert_cleaned_up(&mut new_db);
407}
407}
408
408
@@ -502,7 +502,7 @@ fn alter_column_default() {
502 .unwrap();
502 .unwrap();
503 assert_eq!("NEW DEFAULT", result.get::<_, &str>("name"));
503 assert_eq!("NEW DEFAULT", result.get::<_, &str>("name"));
504
504
505− reshape.complete_migration().unwrap();
505+ reshape.complete().unwrap();
506 common::assert_cleaned_up(&mut new_db);
506 common::assert_cleaned_up(&mut new_db);
507}
507}
508
508
@@ -568,7 +568,7 @@ fn alter_column_with_index() {
568
568
569 // Complete the second migration which should replace the existing column
569 // Complete the second migration which should replace the existing column
570 // with the temporary one
570 // with the temporary one
571− reshape.complete_migration().unwrap();
571+ reshape.complete().unwrap();
572
572
573 // Make sure index still exists
573 // Make sure index still exists
574 let result: i64 = db
574 let result: i64 = db
tests/remove_column.rs+1−1
@@ -76,6 +76,6 @@ fn remove_column() {
76 result.as_ref().map(|row| row.get("name"))
76 result.as_ref().map(|row| row.get("name"))
77 );
77 );
78
78
79− reshape.complete_migration().unwrap();
79+ reshape.complete().unwrap();
80 common::assert_cleaned_up(&mut new_db);
80 common::assert_cleaned_up(&mut new_db);
81}
81}
tests/remove_index.rs+1−1
@@ -85,7 +85,7 @@ fn remove_index() {
85
85
86 assert_eq!(vec![(true, true)], result);
86 assert_eq!(vec![(true, true)], result);
87
87
88− reshape.complete_migration().unwrap();
88+ reshape.complete().unwrap();
89
89
90 // Ensure index has been removed after the migration is complete
90 // Ensure index has been removed after the migration is complete
91 let count: i64 = db
91 let count: i64 = db
tests/remove_table.rs+1−1
@@ -49,6 +49,6 @@ fn remove_table() {
49 // Ensure the table is not accessible through the new schema
49 // Ensure the table is not accessible through the new schema
50 assert!(new_db.query("SELECT id FROM users", &[]).is_err());
50 assert!(new_db.query("SELECT id FROM users", &[]).is_err());
51
51
52− reshape.complete_migration().unwrap();
52+ reshape.complete().unwrap();
53 common::assert_cleaned_up(&mut new_db);
53 common::assert_cleaned_up(&mut new_db);
54}
54}
tests/rename_table.rs+1−1
@@ -75,6 +75,6 @@ fn rename_table() {
75 assert!(old_db.simple_query("SELECT id FROM customers").is_err());
75 assert!(old_db.simple_query("SELECT id FROM customers").is_err());
76 assert!(new_db.simple_query("SELECT id FROM users").is_err());
76 assert!(new_db.simple_query("SELECT id FROM users").is_err());
77
77
78− reshape.complete_migration().unwrap();
78+ reshape.complete().unwrap();
79 common::assert_cleaned_up(&mut new_db);
79 common::assert_cleaned_up(&mut new_db);
80}
80}