Merge lp:~jderose/dmedia/degrade into lp:dmedia

[Jason Gerard DeRose]

For details, see this bug:

https://bugs.launchpad.net/dmedia/+bug/1090917

Changes include:

 * Adds benchmark-purge-store.py to benchmark how quickly we can purge a store; on my system I can purge around 500 files per second, which is workable, although not as fast as I'd like

 * Adds Dmedia.DowngradeStore(store_id) DBus method, adds same to dmedia-cli

 * Moves implementation of Core.purge_store() to MetaStore.purge_store() (better place for it)

 * Adds MetaStore.downgrade_by_never_verified() method and its test

 * Adds MetaStore.downgrade_by_last_verified() method and its test

 * Adds MetaStore.downgrade_by_store_atime() method and its test

 * Adds MetaStore.downgrade_store() method and its test

 * Adds test for file/last-verified view

 * Adds test for store/atime view

 * Adds file/nonzero view, small tweaks to file/last-verified, file/never-verified

[lp:~jderose/dmedia/degrade]

579. By Jason Gerard DeRose on 2012-12-16

Opps, forgot to add test for file/never-verified view

[Jason Gerard DeRose (jderose)]

Also:

 * Adds test for file/never-verified view

 * Changes core.vigilance() to run all these downgrade checks (this is the turned on by default bit)

[Matteo Ronchetti (mttronchetti)]

I think that this code is good

review: Approve

[James Raymond (jamesmr)]

review: Approve

=== added file 'benchmark-purge-store.py'
--- benchmark-purge-store.py	1970-01-01 00:00:00 +0000
+++ benchmark-purge-store.py	2012-12-16 15:30:25 +0000
@@ -0,0 +1,66 @@
+#!/usr/bin/python3
+
+import time
+import logging
+
+from usercouch.misc import TempCouch
+from microfiber import Database, random_id
+from filestore import DIGEST_BYTES
+from dmedia.util import get_db
+from dmedia.metastore import MetaStore, BufferedSave, TimeDelta
+
+logging.basicConfig(level=logging.DEBUG)
+
+couch = TempCouch()
+env = couch.bootstrap()
+db= get_db(env, True)
+ms = MetaStore(db)
+
+store_id1 = random_id()
+store_id2 = random_id()
+store_id3 = random_id()
+
+count = 5000
+buf = BufferedSave(db, 100)
+print('Saving {} docs...'.format(count))
+for i in range(count):
+    doc = {
+        '_id': random_id(DIGEST_BYTES),
+        'time': time.time(),
+        'type': 'dmedia/file',
+        'origin': 'user',
+        'atime': int(time.time()),
+        'bytes': 12345678,
+        'stored': {
+            store_id1: {
+                'copies': 1,
+                'mtime': int(time.time()),
+            },
+            store_id2: {
+                'copies': 2,
+                'mtime': int(time.time()),
+            },
+            store_id3: {
+                'copies': 1,
+                'mtime': int(time.time()),
+            },
+        },
+    }
+    buf.save(doc)
+
+# Prep the view
+db.view('file', 'stored', limit=1)
+
+t = TimeDelta()
+
+#ms.downgrade_store(store_id1)
+#ms.downgrade_store(store_id2)
+#ms.downgrade_store(store_id3)
+
+ms.purge_store(store_id1)
+ms.purge_store(store_id2)
+ms.purge_store(store_id3)
+
+print('Rate: {} per second'.format(count * 3 // t.delta))
+print('')
+

=== modified file 'dmedia-cli'
--- dmedia-cli	2012-11-26 00:04:29 +0000
+++ dmedia-cli	2012-12-16 15:30:25 +0000
@@ -134,6 +134,12 @@
         return [path.abspath(directory)]
 
 
+class DowngradeStore(_Method):
+    'Downgrade durability confidence to zero copies'
+
+    args = ['store_id']
+
+
 class PurgeStore(_Method):
     'Purge references to a store'
 

=== modified file 'dmedia-service'
--- dmedia-service	2012-12-14 05:31:06 +0000
+++ dmedia-service	2012-12-16 15:30:25 +0000
@@ -446,6 +446,12 @@
         return self.LocalDmedia()
 
     @dbus.service.method(IFACE, in_signature='s', out_signature='')
+    def DowngradeStore(self, store_id):
+        store_id = str(store_id)
+        log.info('Dmedia.DowngradeStore(%r)', store_id)
+        start_thread(self.core.downgrade_store, store_id)
+
+    @dbus.service.method(IFACE, in_signature='s', out_signature='')
     def PurgeStore(self, store_id):
         store_id = str(store_id)
         log.info('Dmedia.PurgeStore(%r)', store_id)

=== modified file 'dmedia/core.py'
--- dmedia/core.py	2012-12-14 06:41:01 +0000
+++ dmedia/core.py	2012-12-16 15:30:25 +0000
@@ -257,10 +257,13 @@
             filestores.append(fs)
         for fs in filestores:
             ms.scan(fs)
+        ms.downgrade_by_store_atime()
         for fs in filestores:
             ms.relink(fs)
         for fs in filestores:
             ms.verify_all(fs)
+        ms.downgrade_by_never_verified()
+        ms.downgrade_by_last_verified()
         log.info('vigilance() is exiting...')
     except Exception:
         log.exception('Error in vigilance()')
@@ -482,6 +485,7 @@
         Note that this method makes sense for remote cloud stores as well as for
         local file-stores.
         """
+        return self.ms.downgrade_store(store_id)
 
     def purge_store(self, store_id):
         """
@@ -509,26 +513,7 @@
         Note that this method makes sense for remote cloud stores as well as for
         local file-stores
         """
-        log.info('Purging store %s', store_id)
-        ids = []
-        while True:
-            rows = self.db.view('file', 'stored',
-                key=store_id,
-                include_docs=True,
-                limit=25,
-            )['rows']
-            if not rows:
-                break
-            ids.extend(r['id'] for r in rows)
-            docs = [r['doc'] for r in rows]
-            for doc in docs:
-                del doc['stored'][store_id]
-            try:
-                self.db.save_many(docs)
-            except BulkConflict:
-                log.exception('Conflict purging %s', store_id)
-        log.info('Purged %d references to %s', len(ids), store_id)
-        return ids
+        return self.ms.purge_store(store_id)
 
     def stat(self, _id):
         doc = self.db.get(_id)

=== modified file 'dmedia/metastore.py'
--- dmedia/metastore.py	2012-12-14 08:04:04 +0000
+++ dmedia/metastore.py	2012-12-16 15:30:25 +0000
@@ -47,8 +47,13 @@
 from .util import get_db
 
 
-ONE_WEEK = 60 * 60 * 24 * 7
 log = logging.getLogger()
+DAY = 24 * 60 * 60
+ONE_WEEK = 7 * DAY
+
+DOWNGRADE_BY_STORE_ATIME = 7 * DAY  # 1 week
+DOWNGRADE_BY_NEVER_VERIFIED = 2 * DAY  # 48 hours
+DOWNGRADE_BY_LAST_VERIFIED = 28 * DAY  # 4 weeks
 
 
 class MTimeMismatch(Exception):
@@ -273,7 +278,6 @@
 
     def flush(self):
         if self.docs:
-            log.info('saving %d docs', len(self.docs))
             self.count += len(self.docs)
             try:
                 self.db.save_many(self.docs)
@@ -309,6 +313,107 @@
         log.info('converted mtime from `float` to `int` for %d docs', buf.count)
         return buf.count
 
+    def downgrade_by_never_verified(self, curtime=None):
+        if curtime is None:
+            curtime = int(time.time())
+        assert isinstance(curtime, int) and curtime >= 0
+        endkey = curtime - DOWNGRADE_BY_NEVER_VERIFIED
+        return self._downgrade_by_verified(endkey, 'never-verified')
+
+    def downgrade_by_last_verified(self, curtime=None):
+        if curtime is None:
+            curtime = int(time.time())
+        assert isinstance(curtime, int) and curtime >= 0
+        endkey = curtime - DOWNGRADE_BY_LAST_VERIFIED
+        return self._downgrade_by_verified(endkey, 'last-verified')
+
+    def _downgrade_by_verified(self, endkey, view):
+        t = TimeDelta()
+        count = 0
+        while True:
+            rows = self.db.view('file', view,
+                endkey=endkey,
+                include_docs=True,
+                limit=100,
+            )['rows']
+            if not rows:
+                break
+            dmap = dict(
+                (row['id'], row['doc']) for row in rows
+            )
+            for row in rows:
+                doc = dmap[row['id']]
+                doc['stored'][row['value']]['copies'] = 0
+            docs = list(dmap.values())
+            count += len(docs)
+            try:
+                self.db.save_many(docs)
+            except BulkConflict as e:
+                log.exception('Conflict in downgrade_by %r', view)
+                count -= len(e.conflicts)
+        t.log('downgraded %d files by %s', count, view)
+        return count
+
+    def downgrade_by_store_atime(self, curtime=None):
+        if curtime is None:
+            curtime = int(time.time())
+        assert isinstance(curtime, int) and curtime >= 0
+        rows = self.db.view('store', 'atime',
+            endkey=(curtime - DOWNGRADE_BY_STORE_ATIME)
+        )['rows']
+        ids = [row['id'] for row in rows]
+        for store_id in ids:
+            self.downgrade_store(store_id)
+        return ids
+
+    def downgrade_store(self, store_id):
+        t = TimeDelta()
+        log.info('Downgrading store %s', store_id)
+        count = 0
+        while True:
+            rows = self.db.view('file', 'nonzero',
+                key=store_id,
+                include_docs=True,
+                limit=100,
+            )['rows']
+            if not rows:
+                break
+            docs = [r['doc'] for r in rows]
+            for doc in docs:
+                doc['stored'][store_id]['copies'] = 0
+            count += len(docs)
+            try:
+                self.db.save_many(docs)
+            except BulkConflict as e:
+                log.exception('Conflict downgrading %s', store_id)
+                count -= len(e.conflicts)
+        t.log('downgraded %d copies in %s', count, store_id)
+        return count
+
+    def purge_store(self, store_id):
+        t = TimeDelta()
+        log.info('Purging store %s', store_id)
+        count = 0
+        while True:
+            rows = self.db.view('file', 'stored',
+                key=store_id,
+                include_docs=True,
+                limit=100,
+            )['rows']
+            if not rows:
+                break
+            docs = [r['doc'] for r in rows]
+            for doc in docs:
+                del doc['stored'][store_id]
+            count += len(docs)
+            try:
+                self.db.save_many(docs)
+            except BulkConflict:
+                log.exception('Conflict purging %s', store_id)
+                count -= len(e.conflicts)
+        t.log('Purged %d copies from %s', count, store_id)
+        return count
+
     def scan(self, fs):
         """
         Make sure files we expect to be in the file-store *fs* actually are.

=== modified file 'dmedia/tests/base.py'
--- dmedia/tests/base.py	2012-11-27 03:20:27 +0000
+++ dmedia/tests/base.py	2012-12-16 15:30:25 +0000
@@ -39,6 +39,7 @@
 from microfiber import random_id
 
 
+MAX_SIZE = LEAF_SIZE * 3
 datadir = path.join(path.dirname(path.abspath(__file__)), 'data')
 random = SystemRandom()
 
@@ -149,7 +150,7 @@
         yield Leaf(index, data)
 
 
-def write_random(fp, max_size=LEAF_SIZE*4):
+def write_random(fp, max_size=MAX_SIZE):
     file_size = random.randint(1, max_size)
     h = Hasher()
     for leaf in random_leaves(file_size):
@@ -159,7 +160,7 @@
     return h.content_hash()
 
 
-def random_file(tmpdir, max_size=LEAF_SIZE*4):
+def random_file(tmpdir, max_size=MAX_SIZE):
     filename = path.join(tmpdir, random_id())
     file_size = random.randint(1, max_size)
     dst_fp = open(filename, 'wb')
@@ -223,7 +224,7 @@
         shutil.copy(src, dst)
         return dst
 
-    def random_batch(self, count, empties=0, max_size=LEAF_SIZE*4):
+    def random_batch(self, count, empties=0, max_size=MAX_SIZE):
         result = list(self.random_file(max_size) for i in range(count))
         result.extend(self.random_empty() for i in range(empties))
         result.sort(key=lambda tup: tup[0].name)
@@ -231,7 +232,7 @@
         batch = Batch(files, sum(f.size for f in files), len(files))
         return (batch, result)
 
-    def random_file(self, max_size=LEAF_SIZE*4):
+    def random_file(self, max_size=MAX_SIZE):
         return random_file(self.dir, max_size)
 
     def random_empty(self):

=== modified file 'dmedia/tests/test_core.py'
--- dmedia/tests/test_core.py	2012-12-14 08:25:06 +0000
+++ dmedia/tests/test_core.py	2012-12-16 15:30:25 +0000
@@ -388,97 +388,6 @@
             inst.disconnect_filestore(fs1.parentdir, fs1.id)
         self.assertEqual(str(cm.exception), repr(fs1.parentdir))
 
-    def test_purge_store(self):
-        store_id1 = random_id()
-        store_id2 = random_id()
-        store_id3 = random_id()
-        inst = core.Core(self.env)
-        db = inst.db
-
-        # Test when empty
-        self.assertEqual(inst.purge_store(store_id1), [])
-
-        docs = [
-            {
-                '_id': random_file_id(),
-                'type': 'dmedia/file',
-                'bytes': 1776,
-                'stored': {
-                    store_id1: {
-                        'copies': 1,
-                        'mtime': 1234567890,
-                    },
-                    store_id2: {
-                        'copies': 2,
-                        'mtime': 1234567891,
-                    },
-                },
-            }
-            for i in range(533)
-        ]
-        ids = [doc['_id'] for doc in docs]
-        ids.sort()
-        db.save_many(docs)
-
-        # Test when store isn't present
-        self.assertEqual(inst.purge_store(store_id3), [])
-        for doc in docs:
-            self.assertEqual(db.get(doc['_id']), doc)
-
-        # Purge one of the stores, make sure the other remains
-        self.assertEqual(inst.purge_store(store_id1), ids)
-        for doc in db.get_many(ids):
-            _id = doc['_id']
-            rev = doc.pop('_rev')
-            self.assertTrue(rev.startswith('2-'))
-            self.assertEqual(
-                doc,
-                {
-                    '_id': _id,
-                    'type': 'dmedia/file',
-                    'bytes': 1776,
-                    'stored': {
-                        store_id2: {
-                            'copies': 2,
-                            'mtime': 1234567891,
-                        },
-                    },
-                }
-            )
-
-        # Purge the other store
-        self.assertEqual(inst.purge_store(store_id2), ids)
-        for doc in db.get_many(ids):
-            _id = doc['_id']
-            rev = doc.pop('_rev')
-            self.assertTrue(rev.startswith('3-'))
-            self.assertEqual(
-                doc,
-                {
-                    '_id': _id,
-                    'type': 'dmedia/file',
-                    'bytes': 1776,
-                    'stored': {},
-                }
-            )
-
-        # Purge both again, make sure no doc changes result:
-        self.assertEqual(inst.purge_store(store_id1), [])
-        self.assertEqual(inst.purge_store(store_id2), [])
-        for doc in db.get_many(ids):
-            _id = doc['_id']
-            rev = doc.pop('_rev')
-            self.assertTrue(rev.startswith('3-'))
-            self.assertEqual(
-                doc,
-                {
-                    '_id': _id,
-                    'type': 'dmedia/file',
-                    'bytes': 1776,
-                    'stored': {},
-                }
-            )
-
     def test_update_atime(self):
         inst = core.Core(self.env)
         _id = random_id()

=== modified file 'dmedia/tests/test_metastore.py'
--- dmedia/tests/test_metastore.py	2012-12-14 08:04:04 +0000
+++ dmedia/tests/test_metastore.py	2012-12-16 15:30:25 +0000
@@ -721,6 +721,512 @@
         # Once more with feeling:
         self.assertEqual(ms.schema_check(), 0)
 
+    def test_downgrade_by_never_verified(self):
+        db = util.get_db(self.env, True)
+        ms = metastore.MetaStore(db)
+
+        # Test when empty
+        self.assertEqual(ms.downgrade_by_never_verified(), 0)
+        curtime = int(time.time())
+        self.assertEqual(ms.downgrade_by_never_verified(curtime), 0)
+
+        # Populate
+        base = curtime - metastore.DOWNGRADE_BY_NEVER_VERIFIED
+        store_id1 = random_id()
+        store_id2 = random_id()
+        docs = []
+        count = 10
+        for i in range(count):
+            doc = {
+                '_id': random_file_id(),
+                'type': 'dmedia/file',
+                'stored': {
+                    store_id1: {
+                        'copies': 1,
+                        'mtime': base + i,
+                    },
+                    store_id2: {
+                        'copies': 1,
+                        'mtime': base + i + count,
+                    },
+                },
+            }
+            docs.append(doc)
+        db.save_many(docs)
+        ids = [doc['_id'] for doc in docs]
+
+        # Test when none should be downgraded
+        self.assertEqual(ms.downgrade_by_never_verified(curtime - 1), 0)
+        for (old, new) in zip(docs, db.get_many(ids)):
+            self.assertEqual(old, new)
+
+        # Test when they all should be downgraded
+        self.assertEqual(ms.downgrade_by_never_verified(curtime + 19), 10)
+        for (i, doc) in enumerate(db.get_many(ids)):
+            rev = doc.pop('_rev')
+            self.assertTrue(rev.startswith('2-'))
+            _id = ids[i]
+            self.assertEqual(doc,
+                {
+                    '_id': _id,
+                    'type': 'dmedia/file',
+                    'stored': {
+                        store_id1: {
+                            'copies': 0,
+                            'mtime': base + i,
+                        },
+                        store_id2: {
+                            'copies': 0,
+                            'mtime': base + i + count,
+                        },
+                    },
+                }
+            )
+
+        # Test when they're all already downgraded
+        docs = db.get_many(ids)
+        self.assertEqual(ms.downgrade_by_never_verified(curtime + 19), 0)
+        for (old, new) in zip(docs, db.get_many(ids)):
+            self.assertEqual(old, new)
+
+        # Test when only one store should be downgraded
+        for doc in docs:
+            doc['stored'][store_id1]['copies'] = 1
+            doc['stored'][store_id2]['copies'] = 1
+        db.save_many(docs)
+        self.assertEqual(ms.downgrade_by_never_verified(curtime + 9), 10)
+        for (i, doc) in enumerate(db.get_many(ids)):
+            rev = doc.pop('_rev')
+            self.assertTrue(rev.startswith('4-'))
+            _id = ids[i]
+            self.assertEqual(doc,
+                {
+                    '_id': _id,
+                    'type': 'dmedia/file',
+                    'stored': {
+                        store_id1: {
+                            'copies': 0,
+                            'mtime': base + i,
+                        },
+                        store_id2: {
+                            'copies': 1,
+                            'mtime': base + i + count,
+                        },
+                    },
+                }
+            )
+
+        # Again, test when they're all already downgraded
+        docs = db.get_many(ids)
+        self.assertEqual(ms.downgrade_by_never_verified(curtime + 9), 0)
+        for (old, new) in zip(docs, db.get_many(ids)):
+            self.assertEqual(old, new)
+
+    def test_downgrade_by_last_verified(self):
+        db = util.get_db(self.env, True)
+        ms = metastore.MetaStore(db)
+
+        # Test when empty
+        self.assertEqual(ms.downgrade_by_last_verified(), 0)
+        curtime = int(time.time())
+        self.assertEqual(ms.downgrade_by_last_verified(curtime), 0)
+
+        # Populate
+        base = curtime - metastore.DOWNGRADE_BY_LAST_VERIFIED
+        store_id1 = random_id()
+        store_id2 = random_id()
+        docs = []
+        count = 10
+        for i in range(count):
+            doc = {
+                '_id': random_file_id(),
+                'type': 'dmedia/file',
+                'stored': {
+                    store_id1: {
+                        'copies': 1,
+                        'verified': base + i,
+                    },
+                    store_id2: {
+                        'copies': 1,
+                        'verified': base + i + count,
+                    },
+                },
+            }
+            docs.append(doc)
+        db.save_many(docs)
+        ids = [doc['_id'] for doc in docs]
+
+        # Test when none should be downgraded
+        self.assertEqual(ms.downgrade_by_last_verified(curtime - 1), 0)
+        for (old, new) in zip(docs, db.get_many(ids)):
+            self.assertEqual(old, new)
+
+        # Test when they all should be downgraded
+        self.assertEqual(ms.downgrade_by_last_verified(curtime + 19), 10)
+        for (i, doc) in enumerate(db.get_many(ids)):
+            rev = doc.pop('_rev')
+            self.assertTrue(rev.startswith('2-'))
+            _id = ids[i]
+            self.assertEqual(doc,
+                {
+                    '_id': _id,
+                    'type': 'dmedia/file',
+                    'stored': {
+                        store_id1: {
+                            'copies': 0,
+                            'verified': base + i,
+                        },
+                        store_id2: {
+                            'copies': 0,
+                            'verified': base + i + count,
+                        },
+                    },
+                }
+            )
+
+        # Test when they're all already downgraded
+        docs = db.get_many(ids)
+        self.assertEqual(ms.downgrade_by_last_verified(curtime + 19), 0)
+        for (old, new) in zip(docs, db.get_many(ids)):
+            self.assertEqual(old, new)
+
+        # Test when only one store should be downgraded
+        for doc in docs:
+            doc['stored'][store_id1]['copies'] = 1
+            doc['stored'][store_id2]['copies'] = 1
+        db.save_many(docs)
+        self.assertEqual(ms.downgrade_by_last_verified(curtime + 9), 10)
+        for (i, doc) in enumerate(db.get_many(ids)):
+            rev = doc.pop('_rev')
+            self.assertTrue(rev.startswith('4-'))
+            _id = ids[i]
+            self.assertEqual(doc,
+                {
+                    '_id': _id,
+                    'type': 'dmedia/file',
+                    'stored': {
+                        store_id1: {
+                            'copies': 0,
+                            'verified': base + i,
+                        },
+                        store_id2: {
+                            'copies': 1,
+                            'verified': base + i + count,
+                        },
+                    },
+                }
+            )
+
+        # Again, test when they're all already downgraded
+        docs = db.get_many(ids)
+        self.assertEqual(ms.downgrade_by_last_verified(curtime + 9), 0)
+        for (old, new) in zip(docs, db.get_many(ids)):
+            self.assertEqual(old, new)
+
+    def test_downgrade_by_store_atime(self):   
+        db = util.get_db(self.env, True)
+
+        class Dummy(metastore.MetaStore):
+            def __init__(self, db):
+                super().__init__(db)
+                self._calls = []
+
+            def downgrade_store(self, store_id):
+                self._calls.append(store_id)
+
+        # Test when empty
+        ms = Dummy(db)
+        self.assertEqual(ms.downgrade_by_store_atime(), [])
+        self.assertEqual(ms._calls, [])
+        curtime = int(time.time())
+        self.assertEqual(ms.downgrade_by_store_atime(curtime), [])
+        self.assertEqual(ms._calls, [])
+
+        # Test when some need to be downgraded
+        base = curtime - metastore.DOWNGRADE_BY_STORE_ATIME
+        docs = []
+        for i in range(8):
+            doc = {
+                '_id': random_id(),
+                'type': 'dmedia/store',
+                'atime': base + i,
+            }
+            docs.append(doc)
+        db.save_many(docs)
+        ids = [doc['_id'] for doc in docs]
+        self.assertEqual(ms.downgrade_by_store_atime(curtime - 1), [])
+        self.assertEqual(ms._calls, [])
+        for i in range(8):
+            expected = ids[:i+1]
+            self.assertEqual(
+                ms.downgrade_by_store_atime(curtime + i),
+                expected
+            )
+            self.assertEqual(ms._calls, expected)
+            ms._calls = []
+
+        # Once more with feeling
+        self.assertEqual(
+            ms.downgrade_by_store_atime(curtime),
+            [ids[0]]
+        )
+        self.assertEqual(ms._calls, [ids[0]])
+
+    def test_downgrade_store(self):    
+        db = util.get_db(self.env, True)
+        ms = metastore.MetaStore(db)
+        store_id1 = random_id()
+        store_id2 = random_id()
+        store_id3 = random_id()
+        self.assertEqual(ms.downgrade_store(store_id1), 0)
+        ids = [random_file_id() for i in range(189)]
+        docs = []
+        for _id in ids:
+            doc = {
+                '_id': _id,
+                'type': 'dmedia/file',
+                'stored': {
+                    store_id1: {
+                        'copies': 1,
+                        'mtime': 123,
+                    },
+                    store_id2: {
+                        'copies': 2,
+                        'mtime': 456,
+                    },
+                },
+            }
+            docs.append(doc)
+        db.save_many(docs)
+
+        # Make sure downgrading an unrelated store causes no change:
+        self.assertEqual(ms.downgrade_store(store_id3), 0)
+        for (old, new) in zip(docs, db.get_many(ids)):
+            self.assertEqual(old, new)
+
+        # Downgrade the first store:
+        self.assertEqual(ms.downgrade_store(store_id1), 189)
+        for (_id, doc) in zip(ids, db.get_many(ids)):
+            rev = doc.pop('_rev')
+            self.assertTrue(rev.startswith('2-'))
+            self.assertEqual(doc,
+                {
+                    '_id': _id,
+                    'type': 'dmedia/file',
+                    'stored': {
+                        store_id1: {
+                            'copies': 0,
+                            'mtime': 123,
+                        },
+                        store_id2: {
+                            'copies': 2,
+                            'mtime': 456,
+                        },
+                    },
+                }
+            )
+
+        # Downgrade the 2nd store:
+        self.assertEqual(ms.downgrade_store(store_id2), 189)
+        for (_id, doc) in zip(ids, db.get_many(ids)):
+            rev = doc.pop('_rev')
+            self.assertTrue(rev.startswith('3-'))
+            self.assertEqual(doc,
+                {
+                    '_id': _id,
+                    'type': 'dmedia/file',
+                    'stored': {
+                        store_id1: {
+                            'copies': 0,
+                            'mtime': 123,
+                        },
+                        store_id2: {
+                            'copies': 0,
+                            'mtime': 456,
+                        },
+                    },
+                }
+            )
+
+        # Make sure downgrading both again causes no change:
+        docs = db.get_many(ids)
+        self.assertEqual(ms.downgrade_store(store_id1), 0)
+        self.assertEqual(ms.downgrade_store(store_id2), 0)
+        for (old, new) in zip(docs, db.get_many(ids)):
+            self.assertEqual(old, new)
+
+        # Test when some already have copies=0:
+        sample = random.sample(ids, 23)
+        docs2 = db.get_many(sample)
+        for doc in docs2:
+            doc['stored'][store_id1]['copies'] = 1
+        db.save_many(docs2)
+        self.assertEqual(ms.downgrade_store(store_id1), 23)
+        for (_id, doc) in zip(ids, db.get_many(ids)):
+            rev = doc.pop('_rev')
+            if _id in sample:
+                self.assertTrue(rev.startswith('5-'))
+            else:
+                self.assertTrue(rev.startswith('3-'))
+            self.assertEqual(doc,
+                {
+                    '_id': _id,
+                    'type': 'dmedia/file',
+                    'stored': {
+                        store_id1: {
+                            'copies': 0,
+                            'mtime': 123,
+                        },
+                        store_id2: {
+                            'copies': 0,
+                            'mtime': 456,
+                        },
+                    },
+                }
+            )
+
+        # Test when some have junk values for copies:
+        sample2 = list(filter(lambda _id: _id not in sample, ids))
+        docs2 = db.get_many(sample2)
+        for (i, doc) in enumerate(docs2):
+            # `False` makes sure the file/nonzero view is using !==
+            junk = ('hello', False)[i % 2 == 0]
+            doc['stored'][store_id2]['copies'] = junk
+        db.save_many(docs2)
+        self.assertEqual(ms.downgrade_store(store_id2), 166)
+        for (_id, doc) in zip(ids, db.get_many(ids)):
+            rev = doc.pop('_rev')
+            self.assertTrue(rev.startswith('5-'))
+            self.assertEqual(doc,
+                {
+                    '_id': _id,
+                    'type': 'dmedia/file',
+                    'stored': {
+                        store_id1: {
+                            'copies': 0,
+                            'mtime': 123,
+                        },
+                        store_id2: {
+                            'copies': 0,
+                            'mtime': 456,
+                        },
+                    },
+                }
+            )
+
+        # Again, make sure downgrading both again causes no change:
+        docs = db.get_many(ids)
+        self.assertEqual(ms.downgrade_store(store_id1), 0)
+        self.assertEqual(ms.downgrade_store(store_id2), 0)
+        for (old, new) in zip(docs, db.get_many(ids)):
+            self.assertEqual(old, new)
+
+    def test_purge_store(self):    
+        db = util.get_db(self.env, True)
+        ms = metastore.MetaStore(db)
+        store_id1 = random_id()
+        store_id2 = random_id()
+        store_id3 = random_id()
+
+        # Test when empty:
+        self.assertEqual(ms.purge_store(store_id1), 0)
+
+        ids = [random_file_id() for i in range(189)]
+        docs = []
+        for _id in ids:
+            doc = {
+                '_id': _id,
+                'type': 'dmedia/file',
+                'stored': {
+                    store_id1: {
+                        'copies': 1,
+                        'mtime': 123,
+                    },
+                    store_id2: {
+                        'copies': 2,
+                        'mtime': 456,
+                    },
+                },
+            }
+            docs.append(doc)
+        db.save_many(docs)
+
+        # Make sure purging an unrelated store causes no change:
+        self.assertEqual(ms.purge_store(store_id3), 0)
+        for (old, new) in zip(docs, db.get_many(ids)):
+            self.assertEqual(old, new)
+
+        # Purge the first store:
+        self.assertEqual(ms.purge_store(store_id1), 189)
+        for (_id, doc) in zip(ids, db.get_many(ids)):
+            rev = doc.pop('_rev')
+            self.assertTrue(rev.startswith('2-'))
+            self.assertEqual(doc,
+                {
+                    '_id': _id,
+                    'type': 'dmedia/file',
+                    'stored': {
+                        store_id2: {
+                            'copies': 2,
+                            'mtime': 456,
+                        },
+                    },
+                }
+            )
+
+        # Purge the 2nd store:
+        self.assertEqual(ms.purge_store(store_id2), 189)
+        for (_id, doc) in zip(ids, db.get_many(ids)):
+            rev = doc.pop('_rev')
+            self.assertTrue(rev.startswith('3-'))
+            self.assertEqual(doc,
+                {
+                    '_id': _id,
+                    'type': 'dmedia/file',
+                    'stored': {},
+                }
+            )
+
+        # Make sure purging both again causes no change:
+        docs = db.get_many(ids)
+        self.assertEqual(ms.purge_store(store_id1), 0)
+        self.assertEqual(ms.purge_store(store_id2), 0)
+        for (old, new) in zip(docs, db.get_many(ids)):
+            self.assertEqual(old, new)
+
+        # Test when some already have been purged:
+        sample = random.sample(ids, 23)
+        docs2 = db.get_many(sample)
+        for doc in docs2:
+            doc['stored'] = {
+                store_id1: {
+                    'copies': 1,
+                    'mtime': 123,
+                },
+            }
+        db.save_many(docs2)
+        self.assertEqual(ms.purge_store(store_id1), 23)
+        for (_id, doc) in zip(ids, db.get_many(ids)):
+            rev = doc.pop('_rev')
+            if _id in sample:
+                self.assertTrue(rev.startswith('5-'))
+            else:
+                self.assertTrue(rev.startswith('3-'))
+            self.assertEqual(doc,
+                {
+                    '_id': _id,
+                    'type': 'dmedia/file',
+                    'stored': {},
+                }
+            )
+
+        # Again, make sure purging both again causes no change:
+        docs = db.get_many(ids)
+        self.assertEqual(ms.purge_store(store_id1), 0)
+        self.assertEqual(ms.purge_store(store_id2), 0)
+        for (old, new) in zip(docs, db.get_many(ids)):
+            self.assertEqual(old, new)
+
     def test_scan(self):
         db = util.get_db(self.env, True)
         ms = metastore.MetaStore(db)

=== modified file 'dmedia/tests/test_views.py'
--- dmedia/tests/test_views.py	2012-12-14 22:55:54 +0000
+++ dmedia/tests/test_views.py	2012-12-16 15:30:25 +0000
@@ -30,6 +30,7 @@
 from microfiber import Database, random_id
 from filestore import DIGEST_BYTES
 
+from dmedia.tests.base import random_file_id
 from dmedia.tests.couch import CouchCase
 from dmedia import util, views
 
@@ -993,6 +994,287 @@
             {'rows': [], 'offset': 0, 'total_rows': 0},
         )
 
+    def test_never_verified(self):
+        db = Database('foo', self.env)
+        db.put(None)
+        design = self.build_view('never-verified')
+        db.save(design)
+        self.assertEqual(
+            db.view('file', 'never-verified'),
+            {'rows': [], 'offset': 0, 'total_rows': 0},
+        )
+
+        # Make sure things are well behaved even when doc['stored'] is missed:
+        id1 = random_file_id()
+        doc1 = {
+            '_id': id1,
+            'type': 'dmedia/file',
+        }
+        db.save(doc1)
+        self.assertEqual(
+            db.view('file', 'never-verified'),
+            {'rows': [], 'offset': 0, 'total_rows': 0},
+        )
+
+        # And when doc['stored'] is empty:
+        doc1['stored'] = {}
+        db.save(doc1)
+        self.assertEqual(
+            db.view('file', 'never-verified'),
+            {'rows': [], 'offset': 0, 'total_rows': 0},
+        )
+
+        # Test when there are 2 stores
+        store_id1 = random_id()
+        store_id2 = random_id()
+        doc1['stored'] = {
+            store_id1: {
+                'copies': 1,
+                'mtime': 1001,
+            },
+            store_id2: {
+                'copies': 1,
+                'mtime': 1003,
+            },
+        }
+        db.save(doc1)
+        self.assertEqual(
+            db.view('file', 'never-verified'),
+            {
+                'offset': 0,
+                'total_rows': 2,
+                'rows': [
+                    {'key': 1001, 'id': id1, 'value': store_id1},
+                    {'key': 1003, 'id': id1, 'value': store_id2},
+                ]
+            }
+        )
+
+        # Test that stores are excluded when copies === 0
+        doc1['stored'][store_id1]['copies'] = 0
+        db.save(doc1)
+        self.assertEqual(
+            db.view('file', 'never-verified'),
+            {
+                'offset': 0,
+                'total_rows': 1,
+                'rows': [
+                    {'key': 1003, 'id': id1, 'value': store_id2},
+                ]
+            }
+        )
+        doc1['stored'][store_id2]['copies'] = 0
+        db.save(doc1)
+        self.assertEqual(
+            db.view('file', 'never-verified'),
+            {'rows': [], 'offset': 0, 'total_rows': 0},
+        )
+
+        # Add another doc
+        id2 = random_file_id()
+        doc2 = {
+            '_id': id2,
+            'type': 'dmedia/file',
+            'stored': {
+                store_id1: {
+                    'copies': 2,
+                    'mtime': 1002,
+                },
+                store_id2: {
+                    'copies': 19,
+                    'mtime': 1004,
+                },
+            }
+        }
+        db.save(doc2)
+        self.assertEqual(
+            db.view('file', 'never-verified'),
+            {
+                'offset': 0,
+                'total_rows': 2,
+                'rows': [
+                    {'key': 1002, 'id': id2, 'value': store_id1},
+                    {'key': 1004, 'id': id2, 'value': store_id2},
+                ]
+            }
+        )
+
+        # Make sure it's filtering with !== 0
+        doc1['stored'][store_id1]['copies'] = '0'
+        doc1['stored'][store_id2]['copies'] = False
+        db.save(doc1)
+        self.assertEqual(
+            db.view('file', 'never-verified'),
+            {
+                'offset': 0,
+                'total_rows': 4,
+                'rows': [
+                    {'key': 1001, 'id': id1, 'value': store_id1},
+                    {'key': 1002, 'id': id2, 'value': store_id1},
+                    {'key': 1003, 'id': id1, 'value': store_id2},
+                    {'key': 1004, 'id': id2, 'value': store_id2},
+                ]
+            }
+        )
+
+        # Make sure verified can't be a number
+        doc1['stored'][store_id1]['verified'] = 123
+        doc2['stored'][store_id1]['verified'] = 456
+        db.save_many([doc1, doc2])
+        self.assertEqual(
+            db.view('file', 'never-verified'),
+            {
+                'offset': 0,
+                'total_rows': 2,
+                'rows': [
+                    {'key': 1003, 'id': id1, 'value': store_id2},
+                    {'key': 1004, 'id': id2, 'value': store_id2},
+                ]
+            }
+        )
+
+        # Make sure doc.type is being checked
+        doc1['type'] = 'dmedia/foo'
+        doc2['type'] = 'dmedia/bar'
+        db.save_many([doc1, doc2])
+        self.assertEqual(
+            db.view('file', 'never-verified'),
+            {'rows': [], 'offset': 0, 'total_rows': 0},
+        )
+
+
+    def test_last_verified(self):
+        db = Database('foo', self.env)
+        db.put(None)
+        design = self.build_view('last-verified')
+        db.save(design)
+        self.assertEqual(
+            db.view('file', 'last-verified'),
+            {'rows': [], 'offset': 0, 'total_rows': 0},
+        )
+
+        # Make sure things are well behaved even when doc['stored'] is missed:
+        id1 = random_file_id()
+        doc1 = {
+            '_id': id1,
+            'type': 'dmedia/file',
+        }
+        db.save(doc1)
+        self.assertEqual(
+            db.view('file', 'last-verified'),
+            {'rows': [], 'offset': 0, 'total_rows': 0},
+        )
+
+        # And when doc['stored'] is empty:
+        doc1['stored'] = {}
+        db.save(doc1)
+        self.assertEqual(
+            db.view('file', 'last-verified'),
+            {'rows': [], 'offset': 0, 'total_rows': 0},
+        )
+
+        # Test when there are 2 stores
+        store_id1 = random_id()
+        store_id2 = random_id()
+        doc1['stored'] = {
+            store_id1: {
+                'copies': 1,
+                'verified': 1001,
+            },
+            store_id2: {
+                'copies': 1,
+                'verified': 1003,
+            },
+        }
+        db.save(doc1)
+        self.assertEqual(
+            db.view('file', 'last-verified'),
+            {
+                'offset': 0,
+                'total_rows': 2,
+                'rows': [
+                    {'key': 1001, 'id': id1, 'value': store_id1},
+                    {'key': 1003, 'id': id1, 'value': store_id2},
+                ]
+            }
+        )
+
+        # Test that stores are excluded when copies === 0
+        doc1['stored'][store_id1]['copies'] = 0
+        db.save(doc1)
+        self.assertEqual(
+            db.view('file', 'last-verified'),
+            {
+                'offset': 0,
+                'total_rows': 1,
+                'rows': [
+                    {'key': 1003, 'id': id1, 'value': store_id2},
+                ]
+            }
+        )
+        doc1['stored'][store_id2]['copies'] = 0
+        db.save(doc1)
+        self.assertEqual(
+            db.view('file', 'last-verified'),
+            {'rows': [], 'offset': 0, 'total_rows': 0},
+        )
+
+        # Add another doc
+        id2 = random_file_id()
+        doc2 = {
+            '_id': id2,
+            'type': 'dmedia/file',
+            'stored': {
+                store_id1: {
+                    'copies': 2,
+                    'verified': 1002,
+                },
+                store_id2: {
+                    'copies': 19,
+                    'verified': 1004,
+                },
+            }
+        }
+        db.save(doc2)
+        self.assertEqual(
+            db.view('file', 'last-verified'),
+            {
+                'offset': 0,
+                'total_rows': 2,
+                'rows': [
+                    {'key': 1002, 'id': id2, 'value': store_id1},
+                    {'key': 1004, 'id': id2, 'value': store_id2},
+                ]
+            }
+        )
+
+        # Make sure it's filtering with !== 0
+        doc1['stored'][store_id1]['copies'] = '0'
+        doc1['stored'][store_id2]['copies'] = False
+        db.save(doc1)
+        self.assertEqual(
+            db.view('file', 'last-verified'),
+            {
+                'offset': 0,
+                'total_rows': 4,
+                'rows': [
+                    {'key': 1001, 'id': id1, 'value': store_id1},
+                    {'key': 1002, 'id': id2, 'value': store_id1},
+                    {'key': 1003, 'id': id1, 'value': store_id2},
+                    {'key': 1004, 'id': id2, 'value': store_id2},
+                ]
+            }
+        )
+
+        # Make sure doc.type is being checked
+        doc1['type'] = 'dmedia/foo'
+        doc2['type'] = 'dmedia/bar'
+        db.save_many([doc1, doc2])
+        self.assertEqual(
+            db.view('file', 'last-verified'),
+            {'rows': [], 'offset': 0, 'total_rows': 0},
+        )
+
     def test_verified(self):
         db = Database('foo', self.env)
         db.put(None)
@@ -1299,6 +1581,100 @@
         )
 
 
+class TestStoreDesign(DesignTestCase):
+    """
+    Test each view function in the _design/store design.
+    """
+    design = views.store_design
+
+    def test_atime(self):
+        db = Database('foo', self.env)
+        db.put(None)
+        design = self.build_view('atime')
+        db.save(design)
+
+        self.assertEqual(
+            db.view('store', 'atime'),
+            {'rows': [], 'offset': 0, 'total_rows': 0},
+        )
+
+        docs = []
+        for i in range(9):
+            doc = {
+                '_id': random_id(),
+                'type': 'dmedia/store',
+                'atime': 100 + i
+            }
+            docs.append(doc)
+        db.save_many(docs)
+        self.assertEqual(
+            db.view('store', 'atime'),
+            {
+                'offset': 0,
+                'total_rows': 9,
+                'rows': [
+                    {'key': doc['atime'], 'id': doc['_id'], 'value': None}
+                    for doc in docs
+                ],
+            },
+        )
+
+        # Test our assumputions about endkey
+        self.assertEqual(
+            db.view('store', 'atime', endkey=99),
+            {'offset': 0, 'total_rows': 9, 'rows': []},
+        )
+        self.assertEqual(
+            db.view('store', 'atime', endkey=100),
+            {
+                'offset': 0,
+                'total_rows': 9,
+                'rows': [
+                    {'key': 100, 'id': docs[0]['_id'], 'value': None},
+                ],
+            },
+        )
+        self.assertEqual(
+            db.view('store', 'atime', endkey=102),
+            {
+                'offset': 0,
+                'total_rows': 9,
+                'rows': [
+                    {'key': 100, 'id': docs[0]['_id'], 'value': None},
+                    {'key': 101, 'id': docs[1]['_id'], 'value': None},
+                    {'key': 102, 'id': docs[2]['_id'], 'value': None},
+                ],
+            },
+        )
+
+        # Test when atime is missing
+        doc = docs[-1]
+        del doc['atime']
+        db.save(doc)
+        self.assertEqual(
+            db.view('store', 'atime', endkey=102),
+            {
+                'offset': 0,
+                'total_rows': 9,
+                'rows': [
+                    {'key': None, 'id': doc['_id'], 'value': None},
+                    {'key': 100, 'id': docs[0]['_id'], 'value': None},
+                    {'key': 101, 'id': docs[1]['_id'], 'value': None},
+                    {'key': 102, 'id': docs[2]['_id'], 'value': None},
+                ],
+            },
+        )
+
+        # Make sure doc.type is being checked
+        for doc in docs:
+            doc['type'] = 'dmedia/other'
+        db.save_many(docs)
+        self.assertEqual(
+            db.view('store', 'atime'),
+            {'rows': [], 'offset': 0, 'total_rows': 0},
+        )
+
+
 class TestJobDesign(DesignTestCase):
     """
     Test each view function in the _design/job design.

=== modified file 'dmedia/views.py'
--- dmedia/views.py	2012-12-15 11:13:52 +0000
+++ dmedia/views.py	2012-12-16 15:30:25 +0000
@@ -86,6 +86,19 @@
 }
 """
 
+file_nonzero = """
+function(doc) {
+    if (doc.type == 'dmedia/file') {
+        var key;
+        for (key in doc.stored) {
+            if (doc.stored[key].copies !== 0) {
+                emit(key, null);
+            }
+        }
+    }
+}
+"""
+
 file_copies = """
 function(doc) {
     if (doc.type == 'dmedia/file' && doc.origin == 'user') {
@@ -146,8 +159,8 @@
         var key, value;
         for (key in doc.stored) {
             value = doc.stored[key];
-            if (value.status == 'new') {
-                emit(value.mtime, null);
+            if (typeof value.verified != 'number' && value.copies !== 0) {
+                emit(value.mtime, key);
             }
         }
     }
@@ -157,9 +170,12 @@
 file_last_verified = """
 function(doc) {
     if (doc.type == 'dmedia/file') {
-        var key;
+        var key, value;
         for (key in doc.stored) {
-            emit(doc.stored[key].verified, null);
+            value = doc.stored[key];
+            if (typeof value.verified == 'number' && value.copies !== 0) {
+                emit(value.verified, key);
+            }
         }
     }
 }
@@ -189,11 +205,12 @@
     '_id': '_design/file',
     'views': {
         'stored': {'map': file_stored, 'reduce': _stats},
+        'nonzero': {'map': file_nonzero},
         'copies': {'map': file_copies},
         'fragile': {'map': file_fragile},
         'reclaimable': {'map': file_reclaimable},
-        'never_verified': {'map': file_never_verified},
-        'last_verified': {'map': file_last_verified},
+        'never-verified': {'map': file_never_verified},
+        'last-verified': {'map': file_last_verified},
         'verified': {'map': file_verified},
         'origin': {'map': file_origin, 'reduce': _stats},
     },