[Algorithms/list] Delimiter: Ensure that listing stops beyond prefix

As proved by the test include din a previous commit, the Delimiter extension
was not stopping the listing for keys whose value was greater than the browsed
prefix.

This commit fixes this behavior, and makes the test pass.

Relates to MD-654

lib/algos/list/delimiter.js

@@ -148,10 +148,18 @@ class Delimiter extends Extension {
     filter(obj) {
         const key = obj.key;
         const value = obj.value;
-        if ((this.prefix && !key.startsWith(this.prefix))
-            || (this.alphabeticalOrder
-                && typeof this.NextMarker === 'string'
-                && key <= this.NextMarker)) {
+        if (this.prefix && !key.startsWith(this.prefix)) {
+            // If the key has overstepped the bounds of the prefix we're
+            // listing, given the flat k/v behavior we're expecting, then let's
+            // stop listing there.
+            if (inc(this.prefix) < key) {
+                return FILTER_END;
+            }
+            return FILTER_SKIP;
+        }
+        if (this.alphabeticalOrder
+            && typeof this.NextMarker === 'string'
+            && key <= this.NextMarker) {
             return FILTER_SKIP;
         }
         if (this.delimiter) {

tests/unit/algos/list/MPU.js

@@ -6,7 +6,7 @@ const MultipartUploads =
 const werelogs = require('werelogs').Logger;
 // eslint-disable-next-line new-cap
 const logger = new werelogs('listMpuTest');
-const performListing = require('../../../utils/performListing');
+const { performListing } = require('../../../utils/performListing');
 describe('Multipart Uploads listing algorithm', () => {
     const splitter = '**';
     const overviewPrefix = `overview${splitter}`;

tests/unit/algos/list/basic.js

@@ -4,7 +4,7 @@ const assert = require('assert');
 const Basic = require('../../../../lib/algos/list/basic').List;
 const Werelogs = require('werelogs').Logger;
 const logger = new Werelogs('listTest');
-const performListing = require('../../../utils/performListing');
+const { performListing } = require('../../../utils/performListing');
 
 class Test {
     constructor(name, input, output) {

tests/unit/algos/list/delimiter.js

@@ -7,7 +7,8 @@ const DelimiterMaster =
     require('../../../../lib/algos/list/delimiterMaster').DelimiterMaster;
 const Werelogs = require('werelogs').Logger;
 const logger = new Werelogs('listTest');
-const performListing = require('../../../utils/performListing');
+const { performListing, performListingExtended } =
+    require('../../../utils/performListing');
 const zpad = require('../../helpers').zpad;
 
 class Test {
@@ -371,4 +372,47 @@ describe('Delimiter listing algorithm', () => {
            res = performListing(d, Delimiter, test.input, logger);
            assert.deepStrictEqual(res, test.output);
        });
+
+    it('Should interrupt listing if greater value than the prefix is found',
+        done => {
+            const data = [
+                { key: '/tata.txt', value },
+                { key: '/toto/titi/doggy', value },
+                { key: '/toto/titi/tata', value },
+                { key: '/toto/titi/tutu', value },
+                { key: '/toto/tutu', value },
+                { key: '/tutu.txt', value },
+            ];
+            const test = new Test('interrupt on greater than prefix', {
+                delimiter: '/',
+                alphabeticalOrder: true,
+                prefix: '/toto/titi/',
+                // This marker simulates the bug from MD-653,
+                // Where a marker is set to a wrong, further-on value, and
+                // the listing was not interrupted despite that.
+                marker: '/toto/titi',
+                maxKeys: 10,
+            }, {
+                CommonPrefixes: [],
+                Contents: [
+                    { key: '/toto/titi/doggy', value },
+                    { key: '/toto/titi/tata', value },
+                    { key: '/toto/titi/tutu', value },
+                ],
+                Delimiter: '/',
+                IsTruncated: false,
+                NextMarker: undefined,
+            });
+
+            const { res, counters } =
+                performListingExtended(data, Delimiter, test.input, logger);
+            assert.deepStrictEqual(res, test.output);
+            assert.strictEqual(counters.accepted, 3,
+                               'Expected 3 accepted entries');
+            assert.strictEqual(counters.skipped, 1,
+                               'Expected 1 skipped entries');
+            assert.strictEqual(counters.ended, 1,
+                               'Expected 1 ended event');
+            return done();
+        });
 });

tests/unit/algos/list/delimiterVersions.js

@@ -5,7 +5,7 @@ const DelimiterVersions =
     require('../../../../lib/algos/list/delimiterVersions').DelimiterVersions;
 const Werelogs = require('werelogs').Logger;
 const logger = new Werelogs('listTest');
-const performListing = require('../../../utils/performListing');
+const { performListing } = require('../../../utils/performListing');
 const zpad = require('../../helpers').zpad;
 
 class Test {

tests/utils/performListing.js

@@ -1,5 +1,34 @@
-module.exports = function performListing(data, Extension, params, logger) {
+const {
+    FILTER_END,
+    FILTER_ACCEPT,
+    FILTER_SKIP } = require('../../lib/algos/list/tools');
+
+function performListingExtended(data, Extension, params, logger) {
     const listing = new Extension(params, logger);
-    data.every(e => listing.filter(e) >= 0);
-    return listing.result();
+    const counters = { skipped: 0, ended: 0, accepted: 0 };
+    data.every(e => {
+        const ret = listing.filter(e);
+        if (ret === FILTER_END) {
+            counters.ended += 1;
+            return false;
+        } else if (ret === FILTER_SKIP) {
+            counters.skipped += 1;
+        } else if (ret === FILTER_ACCEPT) {
+            counters.accepted += 1;
+        } else {
+            throw new Error('Unknown filter return value');
+        }
+        return true;
+    });
+    return { res: listing.result(), counters };
+}
+
+function performListing(data, Extension, params, logger) {
+    const retData = performListingExtended(data, Extension, params, logger);
+    return retData.res;
+}
+
+module.exports = {
+    performListing,
+    performListingExtended,
 };