Rework DB lock handling (#141)

Rework handling of exclusive db access

* sqlite3 Python module provides an implicit transaction management
  functionality when isolation_level argument is not None when passed to
  a connect method;
* sqlite3 itself by default operates in the autocommit mode which can
  be disabled by executing the "BEGIN" SQL command;
  https://www.sqlite.org/lockingv3.html#transaction_control
* BEGIN without EXCLUSIVE does not obtain an exclusive lock for the
  database and the default behavior is to run a transaction as DEFERRED;
  https://www.sqlite.org/lang_transaction.html#immediate

The desired behavior for the framework is holding an exclusive access to
the DB while a Framework object exists and is not closed which requires
maintaining exclusive access across transactions. This is possible to
achieve by using `PRAGMA locking_mode=EXCLUSIVE`.

With this change, it will not be possible to have two Framework objects
with exclusive access to the backend storage.

ops/framework.py

@@ -7,6 +7,7 @@
 import collections.abc
 import keyword
 import weakref
+from datetime import timedelta
 
 
 class Handle:
@@ -328,11 +329,16 @@ def __str__(self):
 
 class SQLiteStorage:
 
+    DB_LOCK_TIMEOUT = timedelta(hours=1)
+
     def __init__(self, filename):
-        self._db = sqlite3.connect(str(filename), isolation_level="EXCLUSIVE")
+        # The isolation_level argument is set to None such that the implicit transaction management behavior of the sqlite3 module is disabled.
+        self._db = sqlite3.connect(str(filename), isolation_level=None, timeout=self.DB_LOCK_TIMEOUT.total_seconds())
         self._setup()
 
     def _setup(self):
+        # Make sure that the database is locked until the connection is closed, not until the transaction ends.
+        self._db.execute("PRAGMA locking_mode=EXCLUSIVE")
         c = self._db.execute("BEGIN")
         c.execute("SELECT count(name) FROM sqlite_master WHERE type='table' AND name='snapshot'")
         if c.fetchone()[0] == 0:

test/test_framework.py

@@ -4,12 +4,13 @@
 import tempfile
 import shutil
 import gc
+import datetime
 
 from pathlib import Path
 
 from ops.framework import (
     Framework, Handle, EventSource, EventsBase, EventBase, Object, PreCommitEvent, CommitEvent,
-    NoSnapshotError, StoredState, StoredList, BoundStoredState, StoredStateData
+    NoSnapshotError, StoredState, StoredList, BoundStoredState, StoredStateData, SQLiteStorage
 )
 
 
@@ -18,6 +19,12 @@ class TestFramework(unittest.TestCase):
     def setUp(self):
         self.tmpdir = Path(tempfile.mkdtemp())
         self.addCleanup(shutil.rmtree, self.tmpdir)
+        default_timeout = SQLiteStorage.DB_LOCK_TIMEOUT
+
+        def timeout_cleanup():
+            SQLiteStorage.DB_LOCK_TIMEOUT = default_timeout
+        SQLiteStorage.DB_LOCK_TIMEOUT = datetime.timedelta(0)
+        self.addCleanup(timeout_cleanup)
 
     def create_framework(self):
         framework = Framework(self.tmpdir / "framework.data", self.tmpdir, None, None)
@@ -83,6 +90,7 @@ def restore(self, snapshot):
         framework1.register_type(Foo, None, handle.kind)
         framework1.save_snapshot(event)
         framework1.commit()
+        framework1.close()
 
         framework2 = self.create_framework()
         framework2.register_type(Foo, None, handle.kind)
@@ -97,12 +105,11 @@ def restore(self, snapshot):
 
         framework2.drop_snapshot(event.handle)
         framework2.commit()
+        framework2.close()
 
         framework3 = self.create_framework()
         framework3.register_type(Foo, None, handle.kind)
 
-        self.assertRaises(NoSnapshotError, framework1.load_snapshot, handle)
-        self.assertRaises(NoSnapshotError, framework2.load_snapshot, handle)
         self.assertRaises(NoSnapshotError, framework3.load_snapshot, handle)
 
     def test_simple_event_observer(self):
@@ -215,6 +222,7 @@ def on_commit(self, event):
         self.assertEqual(obs.state.myinitdata, 41)
         self.assertEqual(obs.state.mydata, 42)
         self.assertTrue(obs.seen, [PreCommitEvent, CommitEvent])
+        framework.close()
 
         other_framework = self.create_framework()
 
@@ -394,6 +402,7 @@ class MyObject(Object):
         gc.collect()
         o3 = MyObject(framework, "path")
         self.assertEqual(o1.handle.path, o3.handle.path)
+        framework.close()
         # Or using a second framework
         framework_copy = self.create_framework()
         o_copy = MyObject(framework_copy, "path")
@@ -431,10 +440,12 @@ def restore(self, value):
         # A loaded object also prevents direct creation of an object
         with self.assertRaises(RuntimeError):
             MyObject(framework, "path")
+        framework.close()
         # But we can create an object, or load a snapshot in a copy of the framework
         framework_copy1 = self.create_framework()
         o_copy1 = MyObject(framework_copy1, "path")
         self.assertEqual(o_copy1.value, "path")
+        framework_copy1.close()
         framework_copy2 = self.create_framework()
         framework_copy2.register_type(MyObject, None, MyObject.handle_kind)
         o_copy2 = framework_copy2.load_snapshot(o_handle)
@@ -718,6 +729,13 @@ def test_helper_properties(self):
         self.assertEqual(my_obj.meta, framework.meta)
         self.assertEqual(my_obj.charm_dir, framework.charm_dir)
 
+    def test_ban_concurrent_frameworks(self):
+        f = self.create_framework()
+        with self.assertRaises(Exception) as cm:
+            self.create_framework()
+        self.assertIn('database is locked', str(cm.exception))
+        f.close()
+
 
 class TestStoredState(unittest.TestCase):
 
@@ -981,6 +999,7 @@ def save_snapshot(self, value):
             op(obj_copy1.state.a, b)
             validate_op(obj_copy1.state.a, expected_res)
             framework.commit()
+            framework.close()
 
             framework_copy = self.create_framework(cls=WrappedFramework)
 
@@ -992,6 +1011,7 @@ def save_snapshot(self, value):
             framework.snapshots.clear()
             framework_copy.commit()
             self.assertEqual(framework_copy.snapshots, [])
+            framework_copy.close()
 
     def test_comparison_operations(self):
         test_operations = [(