--- a/bitten/master.py
+++ b/bitten/master.py
@@ -46,6 +46,10 @@
 
         # path to generated snapshot archives, key is (config name, revision)
         self.snapshots = {}
+        for config in BuildConfig.select(self.env):
+            snapshots = archive.index(self.env, prefix=config.name)
+            for rev, format, path in snapshots:
+                self.snapshots[(config.name, rev, format)] = path
 
         self.schedule(self.TRIGGER_INTERVAL, self._check_build_triggers)
 
@@ -55,7 +59,7 @@
             build.delete()
         beep.Listener.close(self)
 
-    def _check_build_triggers(self, master, when):
+    def _check_build_triggers(self, when):
         self.schedule(self.TRIGGER_INTERVAL, self._check_build_triggers)
 
         repos = self.env.get_repository()
@@ -79,7 +83,7 @@
                                          rev, platform.name)
                             build = Build(self.env)
                             build.config = config.name
-                            build.rev = rev
+                            build.rev = str(rev)
                             build.rev_time = repos.get_changeset(rev).date
                             build.platform = platform.id
                             build.insert()
@@ -90,8 +94,9 @@
             repos.close()
 
         self.schedule(5, self._check_build_queue)
+        self.schedule(60, self._cleanup_snapshots)
 
-    def _check_build_queue(self, master, when):
+    def _check_build_queue(self, when):
         if not self.slaves:
             return
         logging.debug('Checking for pending builds...')
@@ -103,21 +108,28 @@
                     slave.send_initiation(build)
                     return
 
-    def get_snapshot(self, build, type, encoding):
-        formats = {
-            ('application/tar', 'bzip2'): 'bzip2',
-            ('application/tar', 'gzip'): 'bzip',
-            ('application/tar', None): 'tar',
-            ('application/zip', None): 'zip',
-        }
-        if not (build.config, build.rev, type, encoding) in self.snapshots:
+    def _cleanup_snapshots(self, when):
+        logging.debug('Checking for unused snapshot archives...')
+        for (config, rev, format), path in self.snapshots.items():
+            keep = False
+            for build in Build.select(self.env, config=config, rev=rev):
+                if build.status not in (Build.SUCCESS, Build.FAILURE):
+                    keep = True
+                    break
+            if not keep:
+                logging.info('Removing unused snapshot %s', path)
+                os.unlink(path)
+                del self.snapshots[(config, rev, format)]
+
+    def get_snapshot(self, build, format):
+        snapshot = self.snapshots.get((build.config, build.rev, format))
+        if not snapshot:
             config = BuildConfig(self.env, build.config)
             snapshot = archive.pack(self.env, path=config.path, rev=build.rev,
-                                    prefix=config.name,
-                                    format=formats[(type, encoding)])
+                                    prefix=config.name, format=format)
             logging.info('Prepared snapshot archive at %s' % snapshot)
-            self.snapshots[(build.config, build.rev, type, encoding)] = snapshot
-        return self.snapshots[(build.config, build.rev, type, encoding)]
+            self.snapshots[(build.config, build.rev, format)] = snapshot
+        return snapshot
 
     def register(self, handler):
         any_match = False
@@ -304,9 +316,13 @@
 
                 build.update()
 
-        # TODO: should not block while reading the file; rather stream it using
-        #       asyncore push_with_producer()
-        snapshot_path = self.master.get_snapshot(build, type, encoding)
+        snapshot_format = {
+            ('application/tar', 'bzip2'): 'bzip2',
+            ('application/tar', 'gzip'): 'gzip',
+            ('application/tar', None): 'tar',
+            ('application/zip', None): 'zip',
+        }[(type, encoding)]
+        snapshot_path = self.master.get_snapshot(build, snapshot_format)
         snapshot_name = os.path.basename(snapshot_path)
         message = beep.Payload(file(snapshot_path), content_type=type,
                                content_disposition=snapshot_name,

--- a/bitten/util/archive.py
+++ b/bitten/util/archive.py
@@ -27,8 +27,31 @@
 _formats = {'gzip': ('.tar.gz', 'gz'), 'bzip2': ('.tar.bz2', 'bz2'),
             'zip': ('.zip', None)}
 
+def index(env, prefix):
+    """Generator that yields `(rev, format, path)` tuples for every archive in
+    the environment snapshots directory that match the specified prefix."""
+    filedir = os.path.join(env.path, 'snapshots')
+    for filename in [f for f in os.listdir(filedir) if f.startswith(prefix)]:
+        rest = filename[len(prefix):]
+
+        # Determine format based of file extension
+        format = None
+        for fmt, (ext, comp) in _formats.items():
+            if rest.endswith(ext):
+                rest = rest[:-len(ext)]
+                format = fmt
+        if not format:
+            continue
+
+        if not rest.startswith('_r'):
+            continue
+        rev = rest[2:]
+
+        yield rev, format, os.path.join(filedir, filename)
+
 def pack(env, repos=None, path=None, rev=None, prefix=None, format='gzip',
          overwrite=False):
+    """Create a snapshot archive in the specified format."""
     if repos is None:
         repos = env.get_repository()
     root = repos.get_node(path or '/', rev)
@@ -79,6 +102,7 @@
     return filename
 
 def unpack(filename, dest_path, format=None):
+    """Extract the contents of a snapshot archive."""
     if not format:
         for name, (extension, compression) in _formats.items():
             if filename.endswith(extension):

--- a/bitten/util/beep.py
+++ b/bitten/util/beep.py
@@ -118,7 +118,7 @@
                 if fired:
                     self.eventqueue = self.eventqueue[j:]
                     for callback in fired:
-                        callback(self, now)
+                        callback(now)
             asyncore.poll(timeout)
 
     def schedule(self, delta, callback):
@@ -134,7 +134,7 @@
         if not self.sessions:
             self.close()
             return
-        def terminate_next_session(session=self, when=None):
+        def terminate_next_session(when):
             session = self.sessions[-1]
             def handle_ok():
                 if self.sessions: