Database changes of resources need to be atomic

deb/local.go

@@ -99,8 +99,8 @@ type LocalRepoCollection struct {
 // NewLocalRepoCollection loads LocalRepos from DB and makes up collection
 func NewLocalRepoCollection(db database.Storage) *LocalRepoCollection {
 	return &LocalRepoCollection{
-		db:      db,
+		db:    db,
-		cache:   make(map[string]*LocalRepo),
+		cache: make(map[string]*LocalRepo),
 	}
 }
 
@@ -158,23 +158,18 @@ func (collection *LocalRepoCollection) Add(repo *LocalRepo) error {
 
 // Update stores updated information about repo in DB
 func (collection *LocalRepoCollection) Update(repo *LocalRepo) error {
-	transaction, err := collection.db.OpenTransaction()
+	batch := collection.db.CreateBatch()
-	if err != nil {
+	err := batch.Put(repo.Key(), repo.Encode())
-		return err
-	}
-	defer transaction.Discard()
-
-	err = transaction.Put(repo.Key(), repo.Encode())
 	if err != nil {
 		return err
 	}
 	if repo.packageRefs != nil {
-		err = transaction.Put(repo.RefKey(), repo.packageRefs.Encode())
+		err = batch.Put(repo.RefKey(), repo.packageRefs.Encode())
 		if err != nil {
 			return err
 		}
 	}
-	return transaction.Commit()
+	return batch.Write()
 }
 
 // LoadComplete loads additional information for local repo
@@ -256,20 +251,16 @@ func (collection *LocalRepoCollection) Drop(repo *LocalRepo) error {
 
 	delete(collection.cache, repo.UUID)
 
-	if _, err = transaction.Get(repo.Key()); err != nil {
+	batch := collection.db.CreateBatch()
-		if err == database.ErrNotFound {
+	err = batch.Delete(repo.Key())
-			return errors.New("local repo not found")
+	if err != nil {
-		}
 		return err
 	}
 
-	if err = transaction.Delete(repo.Key()); err != nil {
+	err = batch.Delete(repo.RefKey())
+	if err != nil {
 		return err
 	}
 
-	if err = transaction.Delete(repo.RefKey()); err != nil {
+	return batch.Write()
-		return err
-	}
-
-	return transaction.Commit()
 }

deb/package_collection.go

@@ -279,7 +279,6 @@ func (collection *PackageCollection) UpdateInTransaction(p *Package, transaction
 	}
 
 	p.collection = collection
-
 	return nil
 }
 

deb/publish.go

@@ -885,7 +885,7 @@ type PublishedRepoCollection struct {
 // NewPublishedRepoCollection loads PublishedRepos from DB and makes up collection
 func NewPublishedRepoCollection(db database.Storage) *PublishedRepoCollection {
 	return &PublishedRepoCollection{
-		db:      db,
+		db: db,
 	}
 }
 
@@ -939,26 +939,27 @@ func (collection *PublishedRepoCollection) CheckDuplicate(repo *PublishedRepo) *
 
 // Update stores updated information about repo in DB
 func (collection *PublishedRepoCollection) Update(repo *PublishedRepo) error {
-	transaction, err := collection.db.OpenTransaction()
+	batch := collection.db.CreateBatch()
+
+	err := batch.Put(repo.Key(), repo.Encode())
 	if err != nil {
 		return err
 	}
-	defer transaction.Discard()
 
-	err = transaction.Put(repo.Key(), repo.Encode())
+	err = batch.Put(repo.Key(), repo.Encode())
 	if err != nil {
 		return err
 	}
 
 	if repo.SourceKind == SourceLocalRepo {
 		for component, item := range repo.sourceItems {
-			err = transaction.Put(repo.RefKey(component), item.packageRefs.Encode())
+			err = batch.Put(repo.RefKey(component), item.packageRefs.Encode())
 			if err != nil {
 				return err
 			}
 		}
 	}
-	return transaction.Commit()
+	return batch.Write()
 }
 
 // LoadComplete loads additional information for remote repo
@@ -1200,12 +1201,6 @@ func (collection *PublishedRepoCollection) Remove(publishedStorageProvider aptly
 	storage, prefix, distribution string, collectionFactory *CollectionFactory, progress aptly.Progress,
 	force, skipCleanup bool) error {
 
-	transaction, err := collection.db.OpenTransaction()
-	if err != nil {
-		return err
-	}
-	defer transaction.Discard()
-
 	// TODO: load via transaction
 	collection.loadList()
 
@@ -1258,17 +1253,18 @@ func (collection *PublishedRepoCollection) Remove(publishedStorageProvider aptly
 		}
 	}
 
-	err = transaction.Delete(repo.Key())
+	batch := collection.db.CreateBatch()
+	err = batch.Delete(repo.Key())
 	if err != nil {
 		return err
 	}
 
 	for _, component := range repo.Components() {
-		err = transaction.Delete(repo.RefKey(component))
+		err = batch.Delete(repo.RefKey(component))
 		if err != nil {
 			return err
 		}
 	}
 
-	return transaction.Commit()
+	return batch.Write()
 }

deb/remote.go

@@ -758,8 +758,8 @@ type RemoteRepoCollection struct {
 // NewRemoteRepoCollection loads RemoteRepos from DB and makes up collection
 func NewRemoteRepoCollection(db database.Storage) *RemoteRepoCollection {
 	return &RemoteRepoCollection{
-		db:      db,
+		db:    db,
-		cache:   make(map[string]*RemoteRepo),
+		cache: make(map[string]*RemoteRepo),
 	}
 }
 
@@ -817,24 +817,19 @@ func (collection *RemoteRepoCollection) Add(repo *RemoteRepo) error {
 
 // Update stores updated information about repo in DB
 func (collection *RemoteRepoCollection) Update(repo *RemoteRepo) error {
-	transaction, err := collection.db.OpenTransaction()
+	batch := collection.db.CreateBatch()
-	if err != nil {
-		return err
-	}
-	defer transaction.Discard()
 
-	err = transaction.Put(repo.Key(), repo.Encode())
+	err := batch.Put(repo.Key(), repo.Encode())
 	if err != nil {
 		return err
 	}
 	if repo.packageRefs != nil {
-		err = transaction.Put(repo.RefKey(), repo.packageRefs.Encode())
+		err = batch.Put(repo.RefKey(), repo.packageRefs.Encode())
 		if err != nil {
 			return err
 		}
 	}
-
+	return batch.Write()
-	return transaction.Commit()
 }
 
 // LoadComplete loads additional information for remote repo
@@ -923,13 +918,15 @@ func (collection *RemoteRepoCollection) Drop(repo *RemoteRepo) error {
 
 	delete(collection.cache, repo.UUID)
 
-	if err = transaction.Delete(repo.Key()); err != nil {
+	batch := collection.db.CreateBatch()
+	err = batch.Delete(repo.Key())
+	if err != nil {
 		return err
 	}
 
-	if err = transaction.Delete(repo.RefKey()); err != nil {
+	err = batch.Delete(repo.RefKey())
+	if err != nil {
 		return err
 	}
-
+	return batch.Write()
-	return transaction.Commit()
 }

deb/snapshot.go

@@ -196,8 +196,8 @@ type SnapshotCollection struct {
 // NewSnapshotCollection loads Snapshots from DB and makes up collection
 func NewSnapshotCollection(db database.Storage) *SnapshotCollection {
 	return &SnapshotCollection{
-		db:      db,
+		db:    db,
-		cache:   map[string]*Snapshot{},
+		cache: map[string]*Snapshot{},
 	}
 }
 
@@ -219,22 +219,20 @@ func (collection *SnapshotCollection) Add(snapshot *Snapshot) error {
 
 // Update stores updated information about snapshot in DB
 func (collection *SnapshotCollection) Update(snapshot *Snapshot) error {
-	transaction, err := collection.db.OpenTransaction()
+	batch := collection.db.CreateBatch()
-	if err != nil {
-		return err
-	}
-	defer transaction.Discard()
 
-	err = transaction.Put(snapshot.Key(), snapshot.Encode())
+	err := batch.Put(snapshot.Key(), snapshot.Encode())
 	if err != nil {
 		return err
 	}
 	if snapshot.packageRefs != nil {
-		if err = transaction.Put(snapshot.RefKey(), snapshot.packageRefs.Encode()); err != nil {
+		err = batch.Put(snapshot.RefKey(), snapshot.packageRefs.Encode())
+		if err != nil {
 			return err
 		}
 	}
-	return transaction.Commit()
+
+	return batch.Write()
 }
 
 // LoadComplete loads additional information about snapshot
@@ -406,15 +404,18 @@ func (collection *SnapshotCollection) Drop(snapshot *Snapshot) error {
 
 	delete(collection.cache, snapshot.UUID)
 
-	if err = transaction.Delete(snapshot.Key()); err != nil {
+	batch := collection.db.CreateBatch()
+	err = batch.Delete(snapshot.Key())
+	if err != nil {
 		return err
 	}
 
-	if err = transaction.Delete(snapshot.RefKey()); err != nil {
+	err = batch.Delete(snapshot.RefKey())
+	if err != nil {
 		return err
 	}
 
-	return transaction.Commit()
+	return batch.Write()
 }
 
 // Snapshot sorting methods