7c48974 · reshape · git.iko.soy

git.iko.soy/reshape

on main help sign in

Use separate table for completed migrations

7c48974fabianlindfors5 years ago

src/db.rs+5

compare with main

unifiedsplit

@@ -57,6 +57,11 @@ impl Transaction<'_> {

57 self.transaction.commit()?;

58 Ok(())

59 }

60+

61+ pub fn rollback(self) -> anyhow::Result<()> {

62+ self.transaction.rollback()?;

63+ Ok(())

64+ }

60}

65}

62impl Conn for Transaction<'_> {

67impl Conn for Transaction<'_> {

src/lib.rs+6−9

@@ -74,8 +74,8 @@ impl Reshape {

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 = &self.state.current_migration.clone();

77+ let current_migration = state::current_migration(&mut self.db)?;

78− let remaining_migrations = self.state.get_remaining_migrations(migrations)?;

78+ let remaining_migrations = state::remaining_migrations(&mut self.db, migrations)?;

79 if remaining_migrations.is_empty() {

80 println!("No migrations left to apply");

81 return Ok(());

@@ -103,7 +103,7 @@ impl Reshape {

103

104 println!("Applying {} migrations\n", remaining_migrations.len());

105

106− helpers::set_up_helpers(&mut self.db, current_migration)

106+ helpers::set_up_helpers(&mut self.db, &current_migration)

107 .context("failed to set up helpers")?;

108

109 let mut new_schema = Schema::new();

@@ -225,7 +225,7 @@ impl Reshape {

225 };

226

227 // Remove previous migration's schema

228− if let Some(current_migration) = &self.state.current_migration {

228+ if let Some(current_migration) = &state::current_migration(&mut self.db)? {

229 self.db

230 .run(&format!(

231 "DROP SCHEMA IF EXISTS {} CASCADE",

@@ -322,11 +322,8 @@ impl Reshape {

322 helpers::tear_down_helpers(&mut self.db).context("failed to tear down helpers")?;

323

324 self.state

325− .complete()

325+ .complete(&mut self.db)

326 .context("failed to update state as completed")?;

327− self.state

328− .save(&mut self.db)

329− .context("failed to save state after setting as completed")?;

330

327

331 Ok(())

328 Ok(())

332 }

329 }

@@ -392,7 +389,7 @@ impl Reshape {

392

389

393 pub fn remove(&mut self) -> anyhow::Result<()> {

390 pub fn remove(&mut self) -> anyhow::Result<()> {

394 // Remove migration schemas and views

391 // Remove migration schemas and views

395− if let Some(current_migration) = &self.state.current_migration {

392+ if let Some(current_migration) = &state::current_migration(&mut self.db)? {

396 self.db.run(&format!(

393 self.db.run(&format!(

397 "DROP SCHEMA IF EXISTS {} CASCADE",

394 "DROP SCHEMA IF EXISTS {} CASCADE",

398 schema_name_for_migration(current_migration)

395 schema_name_for_migration(current_migration)

src/state.rs+122−36

@@ -8,8 +8,6 @@ use version::version;

8pub struct State {

9 pub version: String,

10 pub status: Status,

11− pub current_migration: Option<String>,

12− pub migrations: Vec<Migration>,

13}

11}

15#[derive(Serialize, Deserialize, Clone, Debug)]

13#[derive(Serialize, Deserialize, Clone, Debug)]

@@ -72,25 +70,25 @@ impl State {

73 let default = Self::default();

71 let default = Self::default();

74 self.status = default.status;

72 self.status = default.status;

75− self.current_migration = default.current_migration;

76− self.migrations = default.migrations;

78 Ok(())

74 Ok(())

79 }

75 }

81 // Complete will change the status from Completing to Idle

77 // Complete will change the status from Completing to Idle

82− pub fn complete(&mut self) -> anyhow::Result<()> {

78+ pub fn complete(&mut self, db: &mut impl Conn) -> anyhow::Result<()> {

83 let current_status = std::mem::replace(&mut self.status, Status::Idle);

79 let current_status = std::mem::replace(&mut self.status, Status::Idle);

84−

85 match current_status {

80 match current_status {

86− Status::Completing { mut migrations, .. } => {

81+ Status::Completing { migrations, .. } => {

87− let target_migration = migrations.last().unwrap().name.to_string();

82+ // Add migrations and update state in a transaction to ensure atomicity

88− self.migrations.append(&mut migrations);

83+ let mut transaction = db.transaction()?;

89− self.current_migration = Some(target_migration);

84+ save_migrations(&mut transaction, &migrations)?;

85+ self.save(&mut transaction)?;

86+ transaction.commit()?;

90 }

87 }

91 _ => {

88 _ => {

92 // Move old status back

89 // Move old status back

93 self.status = current_status;

90 self.status = current_status;

91+

94 return Err(anyhow!(

92 return Err(anyhow!(

95 "couldn't update state to be completed, not in Completing state"

93 "couldn't update state to be completed, not in Completing state"

96 ));

94 ));

@@ -138,35 +136,24 @@ impl State {

138 }

136 }

139 }

137 }

140

138

141− pub fn get_remaining_migrations(

142− &self,

143− new_migrations: impl IntoIterator<Item = Migration>,

144− ) -> anyhow::Result<Vec<Migration>> {

145− let mut new_iter = new_migrations.into_iter();

146−

147− // Ensure the new migration match up with the existing ones

148− for pair in self.migrations.iter().zip(new_iter.by_ref()) {

149− let (existing, ref new) = pair;

150− if existing != new {

151− return Err(anyhow!(

152− "existing migration {} does not match new migration {}",

153− existing.name,

154− new.name

155− ));

156− }

157− }

158−

159− let items: Vec<Migration> = new_iter.collect();

160−

161− // Return the remaining migrations

162− Ok(items)

163− }

164−

165 fn ensure_schema_and_table(db: &mut impl Conn) {

139 fn ensure_schema_and_table(db: &mut impl Conn) {

166 db.run("CREATE SCHEMA IF NOT EXISTS reshape").unwrap();

140 db.run("CREATE SCHEMA IF NOT EXISTS reshape").unwrap();

167

141

168 db.run("CREATE TABLE IF NOT EXISTS reshape.data (key TEXT PRIMARY KEY, value JSONB)")

142 db.run("CREATE TABLE IF NOT EXISTS reshape.data (key TEXT PRIMARY KEY, value JSONB)")

169 .unwrap();

143 .unwrap();

144+

145+ db.run(

146+ "

147+ CREATE TABLE IF NOT EXISTS reshape.migrations (

148+ index INTEGER GENERATED ALWAYS AS IDENTITY PRIMARY KEY,

149+ name TEXT NOT NULL,

150+ description TEXT,

151+ actions JSONB NOT NULL,

152+ completed_at TIMESTAMP DEFAULT NOW()

153+ )

154+ ",

155+ )

156+ .unwrap();

170 }

157 }

171}

158}

172

159

@@ -175,8 +162,107 @@ impl Default for State {

175 State {

162 State {

176 version: version!().to_string(),

163 version: version!().to_string(),

177 status: Status::Idle,

164 status: Status::Idle,

178− current_migration: None,

179− migrations: vec![],

180 }

165 }

181 }

166 }

182}

167}

168+

169+pub fn current_migration(db: &mut dyn Conn) -> anyhow::Result<Option<String>> {

170+ let name: Option<String> = db

171+ .query(

172+ "

173+ SELECT name

174+ FROM reshape.migrations

175+ ORDER BY index DESC

176+ LIMIT 1

177+ ",

178+ )?

179+ .first()

180+ .map(|row| row.get("name"));

181+ Ok(name)

182+}

183+

184+pub fn remaining_migrations(

185+ db: &mut impl Conn,

186+ new_migrations: impl IntoIterator<Item = Migration>,

187+) -> anyhow::Result<Vec<Migration>> {

188+ let mut new_iter = new_migrations.into_iter();

189+

190+ // Ensure the new migrations match up with the existing ones

191+ let mut highest_index: Option<i32> = None;

192+ loop {

193+ let migrations = get_migrations(db, highest_index)?;

194+ if migrations.is_empty() {

195+ break;

196+ }

197+

198+ for (index, existing) in migrations {

199+ highest_index = Some(index);

200+

201+ let new = match new_iter.next() {

202+ Some(migration) => migration,

203+ None => {

204+ return Err(anyhow!(

205+ "existing migration {} doesn't exist in local migrations",

206+ existing

207+ ))

208+ }

209+ };

210+

211+ if existing != new.name {

212+ return Err(anyhow!(

213+ "existing migration {} does not match new migration {}",

214+ existing,

215+ new.name

216+ ));

217+ }

218+ }

219+ }

220+

221+ // Return the remaining migrations

222+ let items: Vec<Migration> = new_iter.collect();

223+ Ok(items)

224+}

225+

226+fn get_migrations(

227+ db: &mut impl Conn,

228+ index_larger_than: Option<i32>,

229+) -> anyhow::Result<Vec<(i32, String)>> {

230+ let rows = if let Some(index_larger_than) = index_larger_than {

231+ db.query_with_params(

232+ "

233+ SELECT index, name

234+ FROM reshape.migrations

235+ WHERE index > $1

236+ ORDER BY index ASC

237+ LIMIT 100

238+ ",

239+ &[&index_larger_than],

240+ )?

241+ } else {

242+ db.query(

243+ "

244+ SELECT index, name

245+ FROM reshape.migrations

246+ LIMIT 100

247+ ",

248+ )?

249+ };

250+

251+ let migrations = rows

252+ .iter()

253+ .map(|row| (row.get("index"), row.get("name")))

254+ .collect();

255+ Ok(migrations)

256+}

257+

258+fn save_migrations(db: &mut impl Conn, migrations: &[Migration]) -> anyhow::Result<()> {

259+ for migration in migrations {

260+ let encoded_actions = serde_json::to_value(&migration.actions)?;

261+ db.query_with_params(

262+ "INSERT INTO reshape.migrations(name, description, actions) VALUES ($1, $2, $3)",

263+ &[&migration.name, &migration.description, &encoded_actions],

264+ )?;

265+ }

266+

267+ Ok(())

268+}

tests/add_column.rs−12

@@ -60,10 +60,6 @@ fn add_column() {

60 // Run first migration, should automatically finish

61 reshape.migrate(first_migrations.clone()).unwrap();

62 assert!(matches!(reshape.state.status, Status::Idle));

63− assert_eq!(

64− Some(&create_users_table.name),

65− reshape.state.current_migration.as_ref()

66− );

68 // Update search paths

64 // Update search paths

69 old_db

65 old_db

@@ -156,10 +152,6 @@ fn add_column_nullable() {

156 // Run first migration, should automatically finish

152 // Run first migration, should automatically finish

157 reshape.migrate(first_migrations.clone()).unwrap();

153 reshape.migrate(first_migrations.clone()).unwrap();

158 assert!(matches!(reshape.state.status, Status::Idle));

154 assert!(matches!(reshape.state.status, Status::Idle));

159− assert_eq!(

160− Some(&create_users_table.name),

161− reshape.state.current_migration.as_ref()

162− );

163

155

164 // Update search paths

156 // Update search paths

165 old_db

157 old_db

@@ -255,10 +247,6 @@ fn add_column_with_default() {

255 // Run first migration, should automatically finish

247 // Run first migration, should automatically finish

256 reshape.migrate(first_migrations.clone()).unwrap();

248 reshape.migrate(first_migrations.clone()).unwrap();

257 assert!(matches!(reshape.state.status, Status::Idle));

249 assert!(matches!(reshape.state.status, Status::Idle));

258− assert_eq!(

259− Some(&create_users_table.name),

260− reshape.state.current_migration.as_ref()

261− );

262

250

263 // Update search paths

251 // Update search paths

264 old_db

252 old_db

tests/alter_column.rs−24

@@ -47,10 +47,6 @@ fn alter_column_data() {

47 // Run first migration, should automatically finish

48 reshape.migrate(first_migrations.clone()).unwrap();

49 assert!(matches!(reshape.state.status, Status::Idle));

50− assert_eq!(

51− Some(&create_users_table.name),

52− reshape.state.current_migration.as_ref()

53− );

55 // Update search paths

51 // Update search paths

56 old_db

52 old_db

@@ -151,10 +147,6 @@ fn alter_column_set_not_null() {

151 // Run first migration, should automatically finish

147 // Run first migration, should automatically finish

152 reshape.migrate(first_migrations.clone()).unwrap();

148 reshape.migrate(first_migrations.clone()).unwrap();

153 assert!(matches!(reshape.state.status, Status::Idle));

149 assert!(matches!(reshape.state.status, Status::Idle));

154− assert_eq!(

155− Some(&create_users_table.name),

156− reshape.state.current_migration.as_ref()

157− );

158

150

159 // Update search paths

151 // Update search paths

160 old_db

152 old_db

@@ -256,10 +248,6 @@ fn alter_column_rename() {

256 // Run first migration, should automatically finish

248 // Run first migration, should automatically finish

257 reshape.migrate(first_migrations.clone()).unwrap();

249 reshape.migrate(first_migrations.clone()).unwrap();

258 assert!(matches!(reshape.state.status, Status::Idle));

250 assert!(matches!(reshape.state.status, Status::Idle));

259− assert_eq!(

260− Some(&create_users_table.name),

261− reshape.state.current_migration.as_ref()

262− );

263

251

264 // Update search paths

252 // Update search paths

265 old_db

253 old_db

@@ -356,10 +344,6 @@ fn alter_column_multiple() {

356 // Run first migration, should automatically finish

344 // Run first migration, should automatically finish

357 reshape.migrate(first_migrations.clone()).unwrap();

345 reshape.migrate(first_migrations.clone()).unwrap();

358 assert!(matches!(reshape.state.status, Status::Idle));

346 assert!(matches!(reshape.state.status, Status::Idle));

359− assert_eq!(

360− Some(&create_users_table.name),

361− reshape.state.current_migration.as_ref()

362− );

363

347

364 // Update search paths

348 // Update search paths

365 old_db

349 old_db

@@ -472,10 +456,6 @@ fn alter_column_default() {

472 // Run first migration, should automatically finish

456 // Run first migration, should automatically finish

473 reshape.migrate(first_migrations.clone()).unwrap();

457 reshape.migrate(first_migrations.clone()).unwrap();

474 assert!(matches!(reshape.state.status, Status::Idle));

458 assert!(matches!(reshape.state.status, Status::Idle));

475− assert_eq!(

476− Some(&create_users_table.name),

477− reshape.state.current_migration.as_ref()

478− );

479

459

480 // Update search paths

460 // Update search paths

481 old_db

461 old_db

@@ -585,10 +565,6 @@ fn alter_column_with_index() {

585 // Run first migration, should automatically finish

565 // Run first migration, should automatically finish

586 reshape.migrate(first_migrations.clone()).unwrap();

566 reshape.migrate(first_migrations.clone()).unwrap();

587 assert!(matches!(reshape.state.status, Status::Idle));

567 assert!(matches!(reshape.state.status, Status::Idle));

588− assert_eq!(

589− Some(&create_users_table.name),

590− reshape.state.current_migration.as_ref()

591− );

592

568

593 // Run second migration

569 // Run second migration

594 reshape.migrate(second_migrations.clone()).unwrap();

570 reshape.migrate(second_migrations.clone()).unwrap();

tests/create_table.rs−4

@@ -41,10 +41,6 @@ fn create_table() {

41 .migrate(vec![create_table_migration.clone()])

42 .unwrap();

43 assert!(matches!(reshape.state.status, Status::Idle));

44− assert_eq!(

45− Some(&create_table_migration.name),

46− reshape.state.current_migration.as_ref()

47− );

49 // Ensure table was created

45 // Ensure table was created

50 let result = db

46 let result = db

tests/failure.rs−4

@@ -46,10 +46,6 @@ fn invalid_migration() {

46 // Run first migration, should automatically finish

47 reshape.migrate(first_migrations.clone()).unwrap();

48 assert!(matches!(reshape.state.status, Status::Idle));

49− assert_eq!(

50− Some(&create_users_table.name),

51− reshape.state.current_migration.as_ref()

52− );

54 // Update search paths

50 // Update search paths

55 old_db

51 old_db