Repository: solid/community-server
Pull request #662 (2021-03-10 12:53:32)

feat: redis based locking mechanism


Patch:
diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml
index 3a55d4147..4c1e8cf63 100644
--- a/.github/workflows/ci.yml
+++ b/.github/workflows/ci.yml
@@ -41,6 +41,10 @@ jobs:
           SPARQL_UPDATE: true
         ports:
           - 4000:8890
+      redis:
+        image: redis
+        ports:
+          - 6379:6379
     steps:
       - name: Use Node.js ${{ matrix.node-version }}
         uses: actions/setup-node@v2
diff --git a/package-lock.json b/package-lock.json
index 9db787b9d..e9ffaf138 100644
--- a/package-lock.json
+++ b/package-lock.json
@@ -1152,6 +1152,11 @@
         "@babel/types": "^7.3.0"
       }
     },
+    "@types/bluebird": {
+      "version": "3.5.33",
+      "resolved": "https://registry.npmjs.org/@types/bluebird/-/bluebird-3.5.33.tgz",
+      "integrity": "sha512-ndEo1xvnYeHxm7I/5sF6tBvnsA4Tdi3zj1keRKRs12SP+2ye2A27NDJ1B6PqkfMbGAcT+mqQVqbZRIrhfOp5PQ=="
+    },
     "@types/cookiejar": {
       "version": "2.1.2",
       "resolved": "https://registry.npmjs.org/@types/cookiejar/-/cookiejar-2.1.2.tgz",
@@ -1324,6 +1329,22 @@
         "@types/node": "*"
       }
     },
+    "@types/redis": {
+      "version": "2.8.28",
+      "resolved": "https://registry.npmjs.org/@types/redis/-/redis-2.8.28.tgz",
+      "integrity": "sha512-8l2gr2OQ969ypa7hFOeKqtFoY70XkHxISV0pAwmQ2nm6CSPb1brmTmqJCGGrekCo+pAZyWlNXr+Kvo6L/1wijA==",
+      "requires": {
+        "@types/node": "*"
+      }
+    },
+    "@types/redlock": {
+      "version": "4.0.1",
+      "resolved": "https://registry.npmjs.org/@types/redlock/-/redlock-4.0.1.tgz",
+      "integrity": "sha512-YrPEPHOgLlWtsg/Ocv/gthGDQpx3HXFPhBvCNuBKBKUoMBzvCjCTC95dmJ0WLZgO+fgTxGQFyk6ZO/+/zG5mRg==",
+      "requires": {
+        "@types/bluebird": "*"
+      }
+    },
     "@types/rimraf": {
       "version": "3.0.0",
       "resolved": "https://registry.npmjs.org/@types/rimraf/-/rimraf-3.0.0.tgz",
@@ -2084,6 +2105,11 @@
       "integrity": "sha512-1Yj8h9Q+QDF5FzhMs/c9+6UntbD5MkRfRwac8DoEm9ZfUBZ7tZ55YcGVAzEe4bXsdQHEk+s9S5wsOKVdZrw0tQ==",
       "dev": true
     },
+    "bluebird": {
+      "version": "3.7.2",
+      "resolved": "https://registry.npmjs.org/bluebird/-/bluebird-3.7.2.tgz",
+      "integrity": "sha512-XpNj6GDQzdfW+r2Wnn7xiSAd7TM3jzkxGXBGTtWKuSXv1xUV+azxAm8jdWZN06QTQk+2N2XB9jRDkvbmQmcRtg=="
+    },
     "boxen": {
       "version": "4.2.0",
       "resolved": "https://registry.npmjs.org/boxen/-/boxen-4.2.0.tgz",
@@ -3029,6 +3055,11 @@
       "integrity": "sha1-3zrhmayt+31ECqrgsp4icrJOxhk=",
       "dev": true
     },
+    "denque": {
+      "version": "1.5.0",
+      "resolved": "https://registry.npmjs.org/denque/-/denque-1.5.0.tgz",
+      "integrity": "sha512-CYiCSgIF1p6EUByQPlGkKnP1M9g0ZV3qMIrqMqZqdwazygIA/YP2vrbcyl1h/WppKJTdl1F85cXIle+394iDAQ=="
+    },
     "depd": {
       "version": "1.1.2",
       "resolved": "https://registry.npmjs.org/depd/-/depd-1.1.2.tgz",
@@ -8082,6 +8113,43 @@
         "strip-indent": "^3.0.0"
       }
     },
+    "redis": {
+      "version": "3.0.2",
+      "resolved": "https://registry.npmjs.org/redis/-/redis-3.0.2.tgz",
+      "integrity": "sha512-PNhLCrjU6vKVuMOyFu7oSP296mwBkcE6lrAjruBYG5LgdSqtRBoVQIylrMyVZD/lkF24RSNNatzvYag6HRBHjQ==",
+      "requires": {
+        "denque": "^1.4.1",
+        "redis-commands": "^1.5.0",
+        "redis-errors": "^1.2.0",
+        "redis-parser": "^3.0.0"
+      }
+    },
+    "redis-commands": {
+      "version": "1.7.0",
+      "resolved": "https://registry.npmjs.org/redis-commands/-/redis-commands-1.7.0.tgz",
+      "integrity": "sha512-nJWqw3bTFy21hX/CPKHth6sfhZbdiHP6bTawSgQBlKOVRG7EZkfHbbHwQJnrE4vsQf0CMNE+3gJ4Fmm16vdVlQ=="
+    },
+    "redis-errors": {
+      "version": "1.2.0",
+      "resolved": "https://registry.npmjs.org/redis-errors/-/redis-errors-1.2.0.tgz",
+      "integrity": "sha1-62LSrbFeTq9GEMBK/hUpOEJQq60="
+    },
+    "redis-parser": {
+      "version": "3.0.0",
+      "resolved": "https://registry.npmjs.org/redis-parser/-/redis-parser-3.0.0.tgz",
+      "integrity": "sha1-tm2CjNyv5rS4pCin3vTGvKwxyLQ=",
+      "requires": {
+        "redis-errors": "^1.0.0"
+      }
+    },
+    "redlock": {
+      "version": "4.2.0",
+      "resolved": "https://registry.npmjs.org/redlock/-/redlock-4.2.0.tgz",
+      "integrity": "sha512-j+oQlG+dOwcetUt2WJWttu4CZVeRzUrcVcISFmEmfyuwCVSJ93rDT7YSgg7H7rnxwoRyk/jU46kycVka5tW7jA==",
+      "requires": {
+        "bluebird": "^3.7.2"
+      }
+    },
     "regex-not": {
       "version": "1.0.2",
       "resolved": "https://registry.npmjs.org/regex-not/-/regex-not-1.0.2.tgz",
diff --git a/package.json b/package.json
index 2bb5c2641..cb10d2002 100644
--- a/package.json
+++ b/package.json
@@ -86,6 +86,8 @@
     "@types/pump": "^1.1.0",
     "@types/punycode": "^2.1.0",
     "@types/rdf-js": "^4.0.0",
+    "@types/redis": "^2.8.28",
+    "@types/redlock": "^4.0.1",
     "@types/sparqljs": "^3.1.0",
     "@types/streamify-array": "^1.0.0",
     "@types/uuid": "^8.3.0",
@@ -106,6 +108,8 @@
     "rdf-parse": "^1.7.0",
     "rdf-serialize": "^1.1.0",
     "rdf-terms": "^1.5.1",
+    "redis": "^3.0.2",
+    "redlock": "^4.2.0",
     "sparqlalgebrajs": "^2.5.1",
     "sparqljs": "^3.1.2",
     "streamify-array": "^1.0.1",
diff --git a/src/index.ts b/src/index.ts
index 77a916c75..f4062d5cf 100644
--- a/src/index.ts
+++ b/src/index.ts
@@ -243,6 +243,7 @@ export * from './util/locking/ExpiringReadWriteLocker';
 export * from './util/locking/EqualReadWriteLocker';
 export * from './util/locking/GreedyReadWriteLocker';
 export * from './util/locking/ReadWriteLocker';
+export * from './util/locking/RedisResourceLocker';
 export * from './util/locking/ResourceLocker';
 export * from './util/locking/SingleThreadedResourceLocker';
 export * from './util/locking/WrappedExpiringReadWriteLocker';
diff --git a/src/ldp/http/BasicRequestParser.ts b/src/ldp/http/BasicRequestParser.ts
index b24c8f80d..58ada209d 100644
--- a/src/ldp/http/BasicRequestParser.ts
+++ b/src/ldp/http/BasicRequestParser.ts
@@ -1,4 +1,5 @@
 import type { HttpRequest } from '../../server/HttpRequest';
+import { InternalServerError } from '../../util/errors/InternalServerError';
 import type { Operation } from '../operations/Operation';
 import type { BodyParser } from './BodyParser';
 import type { MetadataExtractor } from './metadata/MetadataExtractor';
@@ -34,7 +35,7 @@ export class BasicRequestParser extends RequestParser {
   public async handle(request: HttpRequest): Promise<Operation> {
     const { method } = request;
     if (!method) {
-      throw new Error('No method specified on the HTTP request');
+      throw new InternalServerError('No method specified on the HTTP request');
     }
     const target = await this.targetExtractor.handleSafe({ request });
     const preferences = await this.preferenceParser.handleSafe({ request });
diff --git a/src/ldp/http/OriginalUrlExtractor.ts b/src/ldp/http/OriginalUrlExtractor.ts
index facf4865a..5e689daf3 100644
--- a/src/ldp/http/OriginalUrlExtractor.ts
+++ b/src/ldp/http/OriginalUrlExtractor.ts
@@ -1,5 +1,7 @@
 import type { TLSSocket } from 'tls';
 import type { HttpRequest } from '../../server/HttpRequest';
+import { BadRequestHttpError } from '../../util/errors/BadRequestHttpError';
+import { InternalServerError } from '../../util/errors/InternalServerError';
 import { parseForwarded } from '../../util/HeaderUtil';
 import { toCanonicalUriPath } from '../../util/PathUtil';
 import type { ResourceIdentifier } from '../representation/ResourceIdentifier';
@@ -18,7 +20,7 @@ export class OriginalUrlExtractor extends TargetExtractor {
 
   public async handle({ request: { url, connection, headers }}: { request: HttpRequest }): Promise<ResourceIdentifier> {
     if (!url) {
-      throw new Error('Missing URL');
+      throw new InternalServerError('Missing URL');
     }
 
     // Extract host and protocol (possibly overridden by the Forwarded/X-Forwarded-* header)
@@ -36,10 +38,10 @@ export class OriginalUrlExtractor extends TargetExtractor {
 
     // Perform a sanity check on the host
     if (!host) {
-      throw new Error('Missing Host header');
+      throw new BadRequestHttpError('Missing Host header');
     }
     if (/[/\\*]/u.test(host)) {
-      throw new Error(`The request has an invalid Host header: ${host}`);
+      throw new BadRequestHttpError(`The request has an invalid Host header: ${host}`);
     }
 
     // URL object applies punycode encoding to domain
diff --git a/src/ldp/representation/RepresentationMetadata.ts b/src/ldp/representation/RepresentationMetadata.ts
index be82e4c7a..a3c495ef5 100644
--- a/src/ldp/representation/RepresentationMetadata.ts
+++ b/src/ldp/representation/RepresentationMetadata.ts
@@ -1,6 +1,7 @@
 import { DataFactory, Store } from 'n3';
 import type { BlankNode, Literal, NamedNode, Quad, Term } from 'rdf-js';
 import { getLoggerFor } from '../../logging/LogUtil';
+import { InternalServerError } from '../../util/errors/InternalServerError';
 import { toSubjectTerm, toObjectTerm, toCachedNamedNode, isTerm } from '../../util/TermUtil';
 import { CONTENT_TYPE, CONTENT_TYPE_TERM } from '../../util/Vocabularies';
 import type { ResourceIdentifier } from './ResourceIdentifier';
@@ -265,7 +266,9 @@ export class RepresentationMetadata {
     }
     if (terms.length > 1) {
       this.logger.error(`Multiple results for ${typeof predicate === 'string' ? predicate : predicate.value}`);
-      throw new Error(`Multiple results for ${typeof predicate === 'string' ? predicate : predicate.value}`);
+      throw new InternalServerError(
+        `Multiple results for ${typeof predicate === 'string' ? predicate : predicate.value}`,
+      );
     }
     return terms[0];
   }
diff --git a/src/server/HttpRequest.ts b/src/server/HttpRequest.ts
index 1ba1dc48b..a071f4d46 100644
--- a/src/server/HttpRequest.ts
+++ b/src/server/HttpRequest.ts
@@ -5,3 +5,10 @@ import type { Guarded } from '../util/GuardedStream';
  * An incoming HTTP request;
  */
 export type HttpRequest = Guarded<IncomingMessage>;
+
+/**
+ * Checks if the given stream is an HttpRequest.
+ */
+export function isHttpRequest(stream: any): stream is HttpRequest {
+  return typeof stream.socket === 'object' && typeof stream.url === 'string' && typeof stream.method === 'string';
+}
diff --git a/src/storage/accessors/InMemoryDataAccessor.ts b/src/storage/accessors/InMemoryDataAccessor.ts
index b885e27a9..e748646b8 100644
--- a/src/storage/accessors/InMemoryDataAccessor.ts
+++ b/src/storage/accessors/InMemoryDataAccessor.ts
@@ -3,6 +3,7 @@ import arrayifyStream from 'arrayify-stream';
 import type { NamedNode } from 'rdf-js';
 import { RepresentationMetadata } from '../../ldp/representation/RepresentationMetadata';
 import type { ResourceIdentifier } from '../../ldp/representation/ResourceIdentifier';
+import { InternalServerError } from '../../util/errors/InternalServerError';
 import { NotFoundHttpError } from '../../util/errors/NotFoundHttpError';
 import type { Guarded } from '../../util/GuardedStream';
 import type { IdentifierStrategy } from '../../util/identifiers/IdentifierStrategy';
@@ -121,7 +122,7 @@ export class InMemoryDataAccessor implements DataAccessor {
         throw new NotFoundHttpError();
       }
       if (this.isDataEntry(parent)) {
-        throw new Error('Invalid path.');
+        throw new InternalServerError('Invalid path.');
       }
     }
 
diff --git a/src/storage/conversion/ChainedConverter.ts b/src/storage/conversion/ChainedConverter.ts
index 83d9d05a8..fc6de1298 100644
--- a/src/storage/conversion/ChainedConverter.ts
+++ b/src/storage/conversion/ChainedConverter.ts
@@ -1,5 +1,6 @@
 import type { Representation } from '../../ldp/representation/Representation';
 import { getLoggerFor } from '../../logging/LogUtil';
+import { InternalServerError } from '../../util/errors/InternalServerError';
 import { matchesMediaType } from './ConversionUtil';
 import type { RepresentationConverterArgs } from './RepresentationConverter';
 import { TypedRepresentationConverter } from './TypedRepresentationConverter';
@@ -79,7 +80,7 @@ export class ChainedConverter extends TypedRepresentationConverter {
 
     if (bestMatch.weight === 0) {
       this.logger.warn(`No match found between ${leftKeys} and ${rightKeys}`);
-      throw new Error(`No match found between ${leftKeys} and ${rightKeys}`);
+      throw new InternalServerError(`No match found between ${leftKeys} and ${rightKeys}`);
     }
 
     this.logger.debug(`${bestMatch.type} is the best match between ${leftKeys} and ${rightKeys}`);
diff --git a/src/storage/mapping/BaseFileIdentifierMapper.ts b/src/storage/mapping/BaseFileIdentifierMapper.ts
index e950b336c..e438842f0 100644
--- a/src/storage/mapping/BaseFileIdentifierMapper.ts
+++ b/src/storage/mapping/BaseFileIdentifierMapper.ts
@@ -2,6 +2,7 @@ import type { ResourceIdentifier } from '../../ldp/representation/ResourceIdenti
 import { getLoggerFor } from '../../logging/LogUtil';
 import { APPLICATION_OCTET_STREAM } from '../../util/ContentTypes';
 import { BadRequestHttpError } from '../../util/errors/BadRequestHttpError';
+import { InternalServerError } from '../../util/errors/InternalServerError';
 import { NotFoundHttpError } from '../../util/errors/NotFoundHttpError';
 import {
   decodeUriPathComponents,
@@ -98,7 +99,7 @@ export class BaseFileIdentifierMapper implements FileIdentifierMapper {
   public async mapFilePathToUrl(filePath: string, isContainer: boolean): Promise<ResourceLink> {
     if (!filePath.startsWith(this.rootFilepath)) {
       this.logger.error(`Trying to access file ${filePath} outside of ${this.rootFilepath}`);
-      throw new Error(`File ${filePath} is not part of the file storage at ${this.rootFilepath}`);
+      throw new InternalServerError(`File ${filePath} is not part of the file storage at ${this.rootFilepath}`);
     }
     const relative = filePath.slice(this.rootFilepath.length);
     let url: string;
diff --git a/src/util/StreamUtil.ts b/src/util/StreamUtil.ts
index 2eb536e70..a0a18030e 100644
--- a/src/util/StreamUtil.ts
+++ b/src/util/StreamUtil.ts
@@ -3,6 +3,7 @@ import { Readable, Transform } from 'stream';
 import arrayifyStream from 'arrayify-stream';
 import pump from 'pump';
 import { getLoggerFor } from '../logging/LogUtil';
+import { isHttpRequest } from '../server/HttpRequest';
 import type { Guarded } from './GuardedStream';
 import { guardStream } from './GuardedStream';
 
@@ -30,14 +31,30 @@ export async function readableToString(stream: Readable): Promise<string> {
  */
 export function pipeSafely<T extends Writable>(readable: NodeJS.ReadableStream, destination: T,
   mapError?: (error: Error) => Error): Guarded<T> {
-  // In case the input readable is guarded, it will no longer log errors since `pump` attaches a new error listener
-  pump(readable, destination, (error): void => {
-    if (error) {
-      logger.warn(`Piped stream errored with ${error.message}`);
-      // Make sure the final error can be handled in a normal streaming fashion
-      destination.emit('error', mapError ? mapError(error) : error);
-    }
-  });
+  // We never want to closes the incoming HttpRequest if there is an error
+  // since that also closes the outgoing HttpResponse.
+  // Since `pump` sends stream errors both up and down the pipe chain,
+  // in this case we need to make sure the error only goes down the chain.
+  if (isHttpRequest(readable)) {
+    readable.pipe(destination);
+    readable.on('error', (error): void => {
+      logger.warn(`HttpRequest errored with ${error.message}`);
+      // From https://nodejs.org/api/stream.html#stream_readable_pipe_destination_options :
+      // One important caveat is that if the Readable stream emits an error during processing,
+      // the Writable destination is not closed automatically. If an error occurs,
+      // it will be necessary to manually close each stream in order to prevent memory leaks.
+      destination.destroy(mapError ? mapError(error) : error);
+    });
+  } else {
+    // In case the input readable is guarded, it will no longer log errors since `pump` attaches a new error listener
+    pump(readable, destination, (error): void => {
+      if (error) {
+        logger.warn(`Piped stream errored with ${error.message}`);
+        // Make sure the final error can be handled in a normal streaming fashion
+        destination.emit('error', mapError ? mapError(error) : error);
+      }
+    });
+  }
   // Guarding the stream now means the internal error listeners of pump will be ignored
   // when checking if there is a valid error listener.
   return guardStream(destination);
diff --git a/src/util/handlers/WaterfallHandler.ts b/src/util/handlers/WaterfallHandler.ts
index 50a5dc5d0..4e7880369 100644
--- a/src/util/handlers/WaterfallHandler.ts
+++ b/src/util/handlers/WaterfallHandler.ts
@@ -48,7 +48,7 @@ export class WaterfallHandler<TIn, TOut> implements AsyncHandler<TIn, TOut> {
       handler = await this.findHandler(input);
     } catch {
       this.logger.warn('All handlers failed. This might be the consequence of calling handle before canHandle.');
-      throw new Error('All handlers failed');
+      throw new InternalServerError('All handlers failed');
     }
 
     return handler.handle(input);
diff --git a/src/util/locking/RedisResourceLocker.ts b/src/util/locking/RedisResourceLocker.ts
new file mode 100644
index 000000000..804d67cd0
--- /dev/null
+++ b/src/util/locking/RedisResourceLocker.ts
@@ -0,0 +1,174 @@
+import { assert } from 'console';
+import type { RedisClient } from 'redis';
+import { createClient } from 'redis';
+import type { Lock } from 'redlock';
+import Redlock from 'redlock';
+import type { ResourceIdentifier } from '../../ldp/representation/ResourceIdentifier';
+import { getLoggerFor } from '../../logging/LogUtil';
+import { InternalServerError } from '../errors/InternalServerError';
+import type { ResourceLocker } from './ResourceLocker';
+
+// The ttl set on a lock, not really important cause redlock wil not handle expiration
+const ttl = 10000;
+// The default redlock config
+const defaultRedlockConfig = {
+  // The expected clock drift; for more details
+  // see http://redis.io/topics/distlock
+  // Multiplied by lock ttl to determine drift time
+  driftFactor: 0.01,
+  // The max number of times Redlock will attempt
+  // to lock a resource before erroring
+  retryCount: 1000000,
+  // The time in ms between attempts
+  retryDelay: 200,
+  // The max time in ms randomly added to retries
+  // to improve performance under high contention
+  // see https://www.awsarchitectureblog.com/2015/03/backoff.html
+  retryJitter: 200,
+};
+/**
+ * A locking system that uses a Redis server or any number of
+ * Redis nodes / clusters
+ * This solution has issues though:
+ *  - Redlock wants to handle expiration itself, this is against the design of a ResourceLocker.
+ *    The workaround for this is to extend an active lock indefinitely.
+ *  - This solution is not multithreaded! If threadA locks a resource, only threadA can unlock this resource.
+ *    If threadB wont be able to lock a resource if threadA already acquired that lock,
+ *    in that sense it is kind of multithreaded.
+ *  - Redlock does not provide the ability to see which locks have expired
+ */
+export class RedisResourceLocker implements ResourceLocker {
+  protected readonly logger = getLoggerFor(this);
+
+  private readonly redlock: Redlock;
+  private readonly lockMap: Map<string, { lock: Lock; interval: NodeJS.Timeout }>;
+
+  public constructor(redisClients: string[], redlockOptions?: Record<string, number>) {
+    this.lockMap = new Map();
+    const clients = this.createRedisClients(redisClients);
+    if (clients.length === 0) {
+      throw new Error('At least 1 client should be provided');
+    }
+    this.redlock = this.createRedlock(clients, redlockOptions);
+    this.redlock.on('clientError', (err): void => {
+      this.logger.error(`Redis/Redlock error: ${err}`);
+    });
+  }
+
+  /**
+   * Generate and return a list of RedisClients based on the provided strings
+   * @param redisClientsStrings - a list of strings that contain either a host address and a
+   * port number like '127.0.0.1:6379' or just a port number like '6379'
+   */
+  private createRedisClients(redisClientsStrings: string[]): RedisClient[] {
+    const result: RedisClient[] = [];
+    if (redisClientsStrings && redisClientsStrings.length > 0) {
+      for (const client of redisClientsStrings) {
+        // Check if port number or ip with port number
+        // Definitely not perfect, but configuring this is only for experienced users
+        const match = new RegExp(/^(?:([^:]+):)?(\d{4,5})$/u, 'u').exec(client);
+        if (!match || !match[2]) {
+          // At least a port number should be provided
+          throw new Error(`Invalid data provided to create a Redis client: ${client}\n
+            Please provide a port number like '6379' or a host address and a port number like '127.0.0.1:6379'`);
+        }
+        const port = Number(match[2]);
+        const host = match[1];
+        const redisclient = createClient(port, host);
+        result.push(redisclient);
+      }
+    }
+    return result;
+  }
+
+  /**
+   * Generate and return a Redlock instance
+   * @param clients - a list of RedisClients you want to use for the redlock instance
+   * @param redlockOptions - extra redlock options to overwrite the default config
+   */
+  private createRedlock(clients: RedisClient[], redlockOptions: Record<string, number> = {}): Redlock {
+    try {
+      return new Redlock(
+        clients,
+        { ...defaultRedlockConfig, ...redlockOptions },
+      );
+    } catch (error: unknown) {
+      throw new InternalServerError(`Error initializing Redlock: ${error}`);
+    }
+  }
+
+  public async quit(): Promise<void> {
+    // This for loop is an extra failsafe,
+    // this extra code won't slow down anything, this function will only be called to shut down in peace
+    for (const [ , { lock }] of this.lockMap.entries()) {
+      await this.release({ path: lock.resource });
+    }
+    await this.redlock.quit();
+  }
+
+  public async acquire(identifier: ResourceIdentifier): Promise<void> {
+    const resource = identifier.path;
+    let lock: Lock | undefined;
+    try {
+      lock = await this.redlock.lock(resource, ttl);
+      assert(lock);
+    } catch (error: unknown) {
+      this.logger.debug(`Unable to acquire lock for ${resource}`);
+      throw new InternalServerError(`Unable to acquire lock for ${resource} (${error})`);
+    }
+    if (this.lockMap.get(resource)) {
+      throw new InternalServerError(`Acquired duplicate lock on ${resource}`);
+    }
+    // Lock acquired
+    this.logger.debug(`Acquired lock for resource: ${resource}!`);
+    const interval = this.extendLockIndefinitely(resource);
+    this.lockMap.set(resource, { lock, interval });
+  }
+
+  public async release(identifier: ResourceIdentifier): Promise<void> {
+    const resource = identifier.path;
+    const entry = this.lockMap.get(resource);
+    if (!entry) {
+      // Lock is invalid
+      this.logger.warn(`Unexpected release request for non-existent lock on ${resource}`);
+      throw new InternalServerError(`Trying to unlock resource that is not locked: ${resource}`);
+    }
+    try {
+      await this.redlock.unlock(entry.lock);
+      clearInterval(entry.interval);
+      this.lockMap.delete(resource);
+      // Successfully released lock
+      this.logger.debug(`Released lock for ${resource}, ${this.getLockCount()} active locks remaining!`);
+    } catch (error: unknown) {
+      this.logger.error(`Error releasing lock for ${resource} (${error})`);
+      throw new InternalServerError(`Unable to release lock for: ${resource}, ${error}`);
+    }
+  }
+
+  /**
+   * Counts the number of active locks.
+   */
+  private getLockCount(): number {
+    return this.lockMap.size;
+  }
+
+  /**
+   * This function is internally used to keep an acquired lock active, a wrapper class will handle expiration
+   * @param identifier - Identifier of the lock to be extended
+   */
+  private extendLockIndefinitely(identifier: string): NodeJS.Timeout {
+    return setInterval(async(): Promise<void> => {
+      const entry = this.lockMap.get(identifier)!;
+      try {
+        const newLock = await this.redlock.extend(entry.lock, ttl);
+        this.lockMap.set(identifier, { lock: newLock, interval: entry.interval });
+        this.logger.debug(`Extended (Redis)lock for resource: ${identifier}`);
+      } catch (error: unknown) {
+        // No error should be re-thrown because this means the lock has simply been released
+        this.logger.error(`Failed to extend this (Redis)lock for resource: ${identifier}, ${error}`);
+        clearInterval(entry.interval);
+        this.lockMap.delete(identifier);
+      }
+    }, ttl / 2);
+  }
+}


Test patch:
diff --git a/test/integration/RedisResourceLockerIntegration.test.ts b/test/integration/RedisResourceLockerIntegration.test.ts
new file mode 100644
index 000000000..6d0859861
--- /dev/null
+++ b/test/integration/RedisResourceLockerIntegration.test.ts
@@ -0,0 +1,173 @@
+import type { Server } from 'http';
+import fetch from 'cross-fetch';
+import type { RedisResourceLocker } from '../../src';
+import { joinFilePath } from '../../src';
+import type { HttpServerFactory } from '../../src/server/HttpServerFactory';
+import { describeIf } from '../util/TestHelpers';
+import { instantiateFromConfig } from './Config';
+/**
+ * Test the general functionality of the server using a RedisResourceLocker
+ */
+describeIf('docker', 'A server with a RedisResourceLocker as ResourceLocker', (): void => {
+  const port = 6008;
+  const baseUrl = `http://localhost:${port}/`;
+  let server: Server;
+  let locker: RedisResourceLocker;
+  let factory: HttpServerFactory;
+
+  beforeAll(async(): Promise<void> => {
+    const instances = await instantiateFromConfig(
+      'urn:solid-server:test:Instances',
+      'run-with-redlock.json',
+      {
+        'urn:solid-server:default:variable:baseUrl': baseUrl,
+        'urn:solid-server:default:variable:podTemplateFolder': joinFilePath(__dirname, '../assets/templates'),
+      },
+    ) as Record<string, any>;
+    ({ factory, locker } = instances);
+    server = factory.startServer(port);
+  });
+
+  afterAll(async(): Promise<void> => {
+    await locker.quit();
+    await new Promise<void>((resolve, reject): void => {
+      server.close((error): void => error ? reject(error) : resolve());
+    });
+  });
+
+  it('can add a file to the store, read it and delete it.', async(): Promise<void> => {
+    // Create file
+    const fileUrl = `${baseUrl}testfile2.txt`;
+    const fileData = 'TESTFILE2';
+
+    let response = await fetch(fileUrl, {
+      method: 'PUT',
+      headers: {
+        'content-type': 'text/plain',
+      },
+      body: fileData,
+    });
+    expect(response.status).toBe(205);
+
+    // Get file
+    response = await fetch(fileUrl);
+    expect(response.status).toBe(200);
+    const body = await response.text();
+    expect(body).toContain('TESTFILE2');
+
+    // DELETE file
+    response = await fetch(fileUrl, { method: 'DELETE' });
+    expect(response.status).toBe(205);
+    response = await fetch(fileUrl);
+    expect(response.status).toBe(404);
+  });
+
+  it('can create a folder and delete it.', async(): Promise<void> => {
+    const containerPath = 'secondfolder/';
+    const containerUrl = `${baseUrl}${containerPath}`;
+    // PUT
+    let response = await fetch(containerUrl, {
+      method: 'PUT',
+      headers: {
+        'content-type': 'text/plain',
+      },
+    });
+    expect(response.status).toBe(205);
+
+    // GET
+    response = await fetch(containerUrl);
+    expect(response.status).toBe(200);
+
+    // DELETE
+    response = await fetch(containerUrl, { method: 'DELETE' });
+    expect(response.status).toBe(205);
+    response = await fetch(containerUrl);
+    expect(response.status).toBe(404);
+  });
+
+  it('can get a resource multiple times.', async(): Promise<void> => {
+    const fileUrl = `${baseUrl}image.png`;
+    const fileData = 'testtesttest';
+
+    let response = await fetch(fileUrl, {
+      method: 'PUT',
+      headers: {
+        'content-type': 'text/plain',
+      },
+      body: fileData,
+    });
+    expect(response.status).toBe(205);
+
+    // GET 4 times
+    for (let i = 0; i < 4; i++) {
+      const res = await fetch(fileUrl);
+      expect(res.status).toBe(200);
+      expect(res.headers.get('content-type')).toBe('text/plain');
+      const body = await res.text();
+      expect(body).toContain('testtesttest');
+    }
+
+    // DELETE
+    response = await fetch(fileUrl, { method: 'DELETE' });
+    expect(response.status).toBe(205);
+    response = await fetch(fileUrl);
+    expect(response.status).toBe(404);
+  });
+
+  describe('Test the ResourceLocker itself', (): void => {
+    const identifier = { path: 'http://test.com/foo' };
+
+    it('can lock and unlock a resource.', async(): Promise<void> => {
+      await expect(locker.acquire(identifier)).resolves.toBeUndefined();
+      await expect(locker.release(identifier)).resolves.toBeUndefined();
+    });
+
+    it('can lock a resource again after it was unlocked.', async(): Promise<void> => {
+      await expect(locker.acquire(identifier)).resolves.toBeUndefined();
+      await expect(locker.release(identifier)).resolves.toBeUndefined();
+      await expect(locker.acquire(identifier)).resolves.toBeUndefined();
+      await expect(locker.release(identifier)).resolves.toBeUndefined();
+    });
+
+    it('can acquire different keys simultaneously.', async(): Promise<void> => {
+      const lock1 = locker.acquire({ path: 'path1' });
+      const lock2 = locker.acquire({ path: 'path2' });
+      const lock3 = locker.acquire({ path: 'path3' });
+
+      await expect(Promise.all([ lock1, lock2, lock3 ])).resolves.toBeDefined();
+
+      await locker.release({ path: 'path1' });
+      await locker.release({ path: 'path2' });
+      await locker.release({ path: 'path3' });
+    });
+
+    it('Cannot acquire the same lock simultaneously.', async(): Promise<void> => {
+      let res = '';
+      const lock1 = locker.acquire(identifier);
+      const lock2 = locker.acquire(identifier);
+      const lock3 = locker.acquire(identifier);
+
+      await new Promise((resolve): any => setImmediate(resolve));
+
+      const l2 = lock2.then(async(): Promise<void> => {
+        res += 'l2';
+        await locker.release(identifier);
+        res += 'r2';
+      });
+      const l1 = lock1.then(async(): Promise<void> => {
+        res += 'l1';
+        await locker.release(identifier);
+        res += 'r1';
+      });
+      const l3 = lock3.then(async(): Promise<void> => {
+        res += 'l3';
+        await locker.release(identifier);
+        res += 'r3';
+      });
+      await Promise.all([ l1, l2, l3 ]);
+      expect(res).toContain('l1r1');
+      expect(res).toContain('l2r2');
+      expect(res).toContain('l3r3');
+    });
+  });
+});
diff --git a/test/integration/ServerFetch.test.ts b/test/integration/ServerFetch.test.ts
index e492468e6..3daa07b0a 100644
--- a/test/integration/ServerFetch.test.ts
+++ b/test/integration/ServerFetch.test.ts
@@ -68,6 +68,18 @@ describe('A Solid server', (): void => {
     expect(res.status).toBe(205);
   });
 
+  it('can handle PUT errors.', async(): Promise<void> => {
+    // There was a specific case where the following request caused the connection to close instead of error
+    const res = await fetch(baseUrl, {
+      method: 'PUT',
+      headers: {
+        'content-type': 'text/plain',
+      },
+      body: '"test"',
+    });
+    expect(res.status).toBe(400);
+  });
+
   it('can POST to create a container.', async(): Promise<void> => {
     const res = await fetch(baseUrl, {
       method: 'POST',
diff --git a/test/integration/config/run-with-redlock.json b/test/integration/config/run-with-redlock.json
new file mode 100644
index 000000000..a982d5205
--- /dev/null
+++ b/test/integration/config/run-with-redlock.json
@@ -0,0 +1,168 @@
+{
+  "@context": "https://linkedsoftwaredependencies.org/bundles/npm/@solid/community-server/^0.0.0/components/context.jsonld",
+  "import": [
+    "files-scs:config/presets/acl.json",
+    "files-scs:config/presets/http.json",
+    "files-scs:config/presets/identifiers/suffix-identifiers.json",
+    "files-scs:config/presets/ldp/credentials-extractor.json",
+    "files-scs:config/presets/ldp/metadata-handler.json",
+    "files-scs:config/presets/ldp/operation-handler.json",
+    "files-scs:config/presets/ldp/permissions-extractor.json",
+    "files-scs:config/presets/ldp/response-writer.json",
+    "files-scs:config/presets/ldp/request-parser.json",
+    "files-scs:config/presets/ldp/websockets.json",
+    "files-scs:config/presets/middleware.json",
+    "files-scs:config/presets/pod-management.json",
+    "files-scs:config/presets/representation-conversion.json",
+    "files-scs:config/presets/static.json",
+    "files-scs:config/presets/storage/backend/storage-memory.json",
+    "files-scs:config/presets/cli-params.json"
+  ],
+  "@graph": [
+    {
+      "@id": "urn:solid-server:test:Instances",
+      "@type": "RecordObject",
+      "RecordObject:_record": [
+        {
+          "RecordObject:_record_key": "factory",
+          "RecordObject:_record_value": {
+            "@id": "urn:solid-server:default:ServerFactory"
+          }
+        },
+        {
+          "RecordObject:_record_key": "locker",
+          "RecordObject:_record_value": {
+            "@id": "urn:solid-server:default:RedisResourceLocker"
+          }
+        }
+      ]
+    },
+    {
+      "@id": "urn:solid-server:default:LdpHandler",
+      "@type": "AuthenticatedLdpHandler",
+      "AuthenticatedLdpHandler:_args_requestParser": {
+        "@id": "urn:solid-server:default:RequestParser"
+      },
+      "AuthenticatedLdpHandler:_args_credentialsExtractor": {
+        "@id": "urn:solid-server:default:CredentialsExtractor"
+      },
+      "AuthenticatedLdpHandler:_args_permissionsExtractor": {
+        "@id": "urn:solid-server:default:PermissionsExtractor"
+      },
+      "AuthenticatedLdpHandler:_args_authorizer": {
+        "@type": "AllowEverythingAuthorizer"
+      },
+      "AuthenticatedLdpHandler:_args_operationHandler": {
+        "@id": "urn:solid-server:default:OperationHandler"
+      },
+      "AuthenticatedLdpHandler:_args_responseWriter": {
+        "@id": "urn:solid-server:default:ResponseWriter"
+      }
+    },
+    {
+      "@id": "urn:solid-server:default:RoutingResourceStore",
+      "@type": "PassthroughStore",
+      "PassthroughStore:_source": {
+        "@id": "urn:solid-server:default:MemoryResourceStore"
+      }
+    },
+
+
+    {
+      "@id": "urn:solid-server:default:ResourceStore",
+      "@type": "MonitoringStore",
+      "MonitoringStore:_source": {
+        "@id": "urn:solid-server:default:ResourceStore_Locking"
+      }
+    },
+
+    {
+      "@id": "urn:solid-server:default:AuxiliaryStrategy",
+      "@type": "RoutingAuxiliaryStrategy",
+      "RoutingAuxiliaryStrategy:_sources": [
+        {
+          "@id": "urn:solid-server:default:AclStrategy"
+        }
+      ]
+    },
+
+    {
+      "@id": "urn:solid-server:default:ResourceStore_Locking",
+      "@type": "LockingResourceStore",
+      "LockingResourceStore:_source": {
+        "@id": "urn:solid-server:default:ResourceStore_Patching"
+      },
+      "LockingResourceStore:_locks": {
+        "@id": "urn:solid-server:default:ResourceLocker"
+      },
+      "LockingResourceStore:_strategy": {
+        "@id": "urn:solid-server:default:AuxiliaryStrategy"
+      }
+    },
+
+    {
+      "@id": "urn:solid-server:default:ResourceStore_Patching",
+      "@type": "PatchingStore",
+      "PatchingStore:_source": {
+        "@id": "urn:solid-server:default:ResourceStore_Converting"
+      },
+      "PatchingStore:_patcher": {
+        "@id": "urn:solid-server:default:PatchHandler",
+        "@type": "SparqlUpdatePatchHandler",
+        "SparqlUpdatePatchHandler:_source": {
+          "@id": "urn:solid-server:default:ResourceStore_ToTurtle"
+        }
+      }
+    },
+
+    {
+      "@id": "urn:solid-server:default:ResourceLocker",
+      "@type": "WrappedExpiringReadWriteLocker",
+      "WrappedExpiringReadWriteLocker:_locker": {
+        "@type": "GreedyReadWriteLocker",
+        "GreedyReadWriteLocker:_locker": {
+          "@id": "urn:solid-server:default:RedisResourceLocker",
+          "@type": "RedisResourceLocker",
+          "RedisResourceLocker:_redisClients": [ "6379" ]
+        },
+        "GreedyReadWriteLocker:_storage": {
+          "@type": "ResourceIdentifierStorage",
+          "ResourceIdentifierStorage:_source": {
+            "@type": "MemoryMapStorage"
+          }
+        },
+        "GreedyReadWriteLocker:_suffixes_count": "count",
+        "GreedyReadWriteLocker:_suffixes_read": "read",
+        "GreedyReadWriteLocker:_suffixes_write": "write"
+      },
+      "WrappedExpiringReadWriteLocker:_expiration": 3000
+    },
+
+
+    {
+      "@id": "urn:solid-server:default:ResourceStore_ToTurtle",
+      "@type": "RepresentationConvertingStore",
+      "RepresentationConvertingStore:_source": {
+        "@id": "urn:solid-server:default:RoutingResourceStore"
+      },
+      "RepresentationConvertingStore:_options_outConverter": {
+        "@id": "urn:solid-server:default:RepresentationConverter"
+      },
+      "RepresentationConvertingStore:_options_inConverter": {
+        "@id": "urn:solid-server:default:RepresentationConverter"
+      },
+      "RepresentationConvertingStore:_options_inType": "text/turtle"
+    },
+
+    {
+      "@id": "urn:solid-server:default:ResourceStore_Converting",
+      "@type": "RepresentationConvertingStore",
+      "RepresentationConvertingStore:_source": {
+        "@id": "urn:solid-server:default:RoutingResourceStore"
+      },
+      "RepresentationConvertingStore:_options_outConverter": {
+        "@id": "urn:solid-server:default:RepresentationConverter"
+      }
+    }
+  ]
+}
diff --git a/test/unit/server/HttpRequest.test.ts b/test/unit/server/HttpRequest.test.ts
new file mode 100644
index 000000000..461bcce85
--- /dev/null
+++ b/test/unit/server/HttpRequest.test.ts
@@ -0,0 +1,10 @@
+import { isHttpRequest } from '../../../src/server/HttpRequest';
+
+describe('HttpRequest', (): void => {
+  describe('#isHttpRequest', (): void => {
+    it('can identify HttpRequests.', async(): Promise<void> => {
+      expect(isHttpRequest({})).toBe(false);
+      expect(isHttpRequest({ socket: {}, method: 'GET', url: '/url' })).toBe(true);
+    });
+  });
+});
diff --git a/test/unit/util/StreamUtil.test.ts b/test/unit/util/StreamUtil.test.ts
index 70bfeb448..be3e0b1d6 100644
--- a/test/unit/util/StreamUtil.test.ts
+++ b/test/unit/util/StreamUtil.test.ts
@@ -1,8 +1,13 @@
 import { PassThrough } from 'stream';
 import arrayifyStream from 'arrayify-stream';
 import streamifyArray from 'streamify-array';
+import { isHttpRequest } from '../../../src/server/HttpRequest';
 import { guardedStreamFrom, pipeSafely, transformSafely, readableToString } from '../../../src/util/StreamUtil';
 
+jest.mock('../../../src/server/HttpRequest', (): any => ({
+  isHttpRequest: jest.fn(),
+}));
+
 describe('StreamUtil', (): void => {
   describe('#readableToString', (): void => {
     it('concatenates all elements of a Readable.', async(): Promise<void> => {
@@ -12,6 +17,10 @@ describe('StreamUtil', (): void => {
   });
 
   describe('#pipeSafely', (): void => {
+    beforeEach(async(): Promise<void> => {
+      (isHttpRequest as unknown as jest.Mock).mockClear();
+    });
+
     it('pipes data from one stream to the other.', async(): Promise<void> => {
       const input = streamifyArray([ 'data' ]);
       const output = new PassThrough();
@@ -56,6 +65,47 @@ describe('StreamUtil', (): void => {
       await new Promise(setImmediate);
       expect(input.destroyed).toBe(true);
     });
+
+    it('does not destroy the source stream if it is an HttpRequest.', async(): Promise<void> => {
+      (isHttpRequest as unknown as jest.Mock).mockReturnValueOnce(true);
+      const input = new PassThrough();
+      const output = new PassThrough();
+      const piped = pipeSafely(input, output);
+
+      // Catch errors to prevent problems in test output
+      output.on('error', (): void => {
+        // Empty
+      });
+
+      piped.destroy(new Error('error!'));
+      // Allow events to propagate
+      await new Promise(setImmediate);
+      expect(input.destroyed).toBe(false);
+    });
+
+    it('still sends errors downstream if the input is an HttpRequest.', async(): Promise<void> => {
+      (isHttpRequest as unknown as jest.Mock).mockReturnValueOnce(true);
+      const input = new PassThrough();
+      input.read = (): any => {
+        input.emit('error', new Error('error'));
+        return null;
+      };
+      const output = new PassThrough();
+      const piped = pipeSafely(input, output);
+      await expect(readableToString(piped)).rejects.toThrow('error');
+    });
+
+    it('can map errors if the input is an HttpRequest.', async(): Promise<void> => {
+      (isHttpRequest as unknown as jest.Mock).mockReturnValueOnce(true);
+      const input = streamifyArray([ 'data' ]);
+      input.read = (): any => {
+        input.emit('error', new Error('error'));
+        return null;
+      };
+      const output = new PassThrough();
+      const piped = pipeSafely(input, output, (): any => new Error('other error'));
+      await expect(readableToString(piped)).rejects.toThrow('other error');
+    });
   });
 
   describe('#transformSafely', (): void => {
@@ -158,8 +208,8 @@ describe('StreamUtil', (): void => {
 
   describe('#guardedStreamFrom', (): void => {
     it('converts data to a guarded stream.', async(): Promise<void> => {
-      const data = [ 'a', 'b' ];
-      await expect(readableToString(guardedStreamFrom(data))).resolves.toBe('ab');
+      await expect(readableToString(guardedStreamFrom([ 'a', 'b' ]))).resolves.toBe('ab');
+      await expect(readableToString(guardedStreamFrom('ab'))).resolves.toBe('ab');
     });
   });
 });
diff --git a/test/unit/util/locking/RedisResourceLocker.test.ts b/test/unit/util/locking/RedisResourceLocker.test.ts
new file mode 100644
index 000000000..dce5d80c5
--- /dev/null
+++ b/test/unit/util/locking/RedisResourceLocker.test.ts
@@ -0,0 +1,208 @@
+import { EventEmitter } from 'events';
+// eslint-disable-next-line import/default
+import redis from 'redis';
+import Redlock from 'redlock';
+import type { Lock } from 'redlock';
+import * as LogUtil from '../../../../src/logging/LogUtil';
+import { InternalServerError } from '../../../../src/util/errors/InternalServerError';
+import { RedisResourceLocker } from '../../../../src/util/locking/RedisResourceLocker';
+
+const redlock: jest.Mocked<Redlock> = Object.assign(new EventEmitter(), {
+  lock: jest.fn().mockImplementation(async(resource: string, ttl: number): Promise<Lock> =>
+    ({ resource, expiration: Date.now() + ttl } as Lock)),
+  unlock: jest.fn(),
+  extend: jest.fn().mockImplementation(
+    async(lock: Lock, ttl: number): Promise<Lock> => {
+      lock.expiration += ttl;
+      return lock;
+    },
+  ),
+  quit: jest.fn(),
+}) as any;
+
+jest.mock('redlock', (): any => jest.fn().mockImplementation((): Redlock => redlock));
+
+jest.useFakeTimers();
+
+describe('A RedisResourceLocker', (): void => {
+  let locker: RedisResourceLocker;
+  const identifier = { path: 'http://test.com/foo' };
+  let createClient: jest.SpyInstance;
+
+  beforeEach(async(): Promise<void> => {
+    jest.clearAllMocks();
+    redlock.removeAllListeners();
+
+    createClient = jest.spyOn(redis, 'createClient').mockImplementation(jest.fn());
+
+    locker = new RedisResourceLocker([ '6379' ]);
+  });
+
+  afterEach(async(): Promise<void> => {
+    // In case some locks are not released by a test the timers will still be running
+    jest.clearAllTimers();
+  });
+
+  afterAll(async(): Promise<void> => {
+    jest.restoreAllMocks();
+  });
+
+  it('can lock and unlock a resource.', async(): Promise<void> => {
+    await expect(locker.acquire(identifier)).resolves.toBeUndefined();
+    await expect(locker.release(identifier)).resolves.toBeUndefined();
+    expect(redlock.lock).toHaveBeenCalledTimes(1);
+    expect(redlock.unlock).toHaveBeenCalledTimes(1);
+  });
+
+  it('can lock a resource again after it was unlocked.', async(): Promise<void> => {
+    await expect(locker.acquire(identifier)).resolves.toBeUndefined();
+    await expect(locker.release(identifier)).resolves.toBeUndefined();
+    await expect(locker.acquire(identifier)).resolves.toBeUndefined();
+    expect(redlock.lock).toHaveBeenCalledTimes(2);
+    expect(redlock.unlock).toHaveBeenCalledTimes(1);
+    await expect(locker.release(identifier)).resolves.toBeUndefined();
+  });
+
+  it('errors when unlocking a resource that was not locked.', async(): Promise<void> => {
+    await expect(locker.acquire(identifier)).resolves.toBeUndefined();
+    await expect(locker.release(identifier)).resolves.toBeUndefined();
+    await expect(locker.release(identifier)).rejects.toThrow(InternalServerError);
+    expect(redlock.lock).toHaveBeenCalledTimes(1);
+    expect(redlock.unlock).toHaveBeenCalledTimes(1);
+  });
+
+  it('errors when redlock.lock throws an error.', async(): Promise<void> => {
+    redlock.lock.mockRejectedValueOnce(new Error('random Error'));
+    const prom = locker.acquire(identifier);
+    await expect(prom).rejects.toThrow(InternalServerError);
+    await expect(prom).rejects.toThrow('Unable to acquire lock for ');
+    await expect(prom).rejects.toThrow('Error: random Error');
+    expect(redlock.lock).toHaveBeenCalledTimes(1);
+  });
+
+  it('errors if redlock.lock resolves but a lock is already stored.', async(): Promise<void> => {
+    await expect(locker.acquire(identifier)).resolves.toBeUndefined();
+    // Works since redlock.lock is mocked to always succeed
+    const prom = locker.acquire(identifier);
+    await expect(prom).rejects.toThrow(InternalServerError);
+    await expect(prom).rejects.toThrow(`Acquired duplicate lock on ${identifier.path}`);
+  });
+
+  it('errors when redlock.unlock throws an error.', async(): Promise<void> => {
+    await locker.acquire(identifier);
+    redlock.unlock.mockRejectedValueOnce(new Error('random Error'));
+    const prom = locker.release(identifier);
+    await expect(prom).rejects.toThrow(InternalServerError);
+    await expect(prom).rejects.toThrow('Unable to release lock for: ');
+    await expect(prom).rejects.toThrow('Error: random Error');
+    expect(redlock.unlock).toHaveBeenCalledTimes(1);
+    await expect(locker.release(identifier)).resolves.toBeUndefined();
+  });
+
+  it('does not extend when there are no locks to extend.', async(): Promise<void> => {
+    await locker.acquire(identifier);
+    await locker.release(identifier);
+    jest.advanceTimersByTime(20000);
+    expect(redlock.extend).toHaveBeenCalledTimes(0);
+  });
+
+  it('cleans up if lock extension failed.', async(): Promise<void> => {
+    // This should never happen though
+    redlock.extend.mockImplementationOnce((): any => {
+      throw new Error('random error');
+    });
+    await locker.acquire(identifier);
+    jest.advanceTimersByTime(20000);
+    expect(redlock.extend).toHaveBeenCalledTimes(1);
+    // Will throw since we removed the lock entry
+    await expect(locker.release(identifier)).rejects.toThrow(InternalServerError);
+  });
+
+  it('can acquire different keys simultaneously.', async(): Promise<void> => {
+    const lock1 = locker.acquire({ path: 'path1' });
+    const lock2 = locker.acquire({ path: 'path2' });
+    const lock3 = locker.acquire({ path: 'path3' });
+
+    await expect(Promise.all([ lock1, lock2, lock3 ])).resolves.toBeDefined();
+
+    await locker.release({ path: 'path1' });
+    await locker.release({ path: 'path2' });
+    await locker.release({ path: 'path3' });
+  });
+
+  it('extends a lock indefinitely.', async(): Promise<void> => {
+    await locker.acquire(identifier);
+    jest.advanceTimersByTime(20000);
+    await expect(locker.release(identifier)).resolves.toBeUndefined();
+  });
+
+  it('uses users redlockOptions if passed to constructor.', async(): Promise<void> => {
+    // Reset calls done in `beforeEach`
+    jest.clearAllMocks();
+    const clients = [ '6379' ];
+    const options = {
+      driftFactor: 0.2,
+      retryCount: 20,
+      retryDelay: 2000,
+      retryJitter: 2000,
+    };
+    locker = new RedisResourceLocker(clients, options);
+    expect(Redlock).toHaveBeenCalledTimes(1);
+    expect(Redlock).toHaveBeenLastCalledWith(expect.any(Array), options);
+  });
+
+  it('errors on creation when no redis servers are passed to the constructor.', async(): Promise<void> => {
+    expect((): any => new RedisResourceLocker([])).toThrow('At least 1 client should be provided');
+  });
+
+  it('errors if there is an issue creating the Redlock.', async(): Promise<void> => {
+    (Redlock as unknown as jest.Mock).mockImplementationOnce((): never => {
+      throw new Error('redlock error!');
+    });
+    expect((): any => new RedisResourceLocker([ '1234' ]))
+      .toThrow('Error initializing Redlock: Error: redlock error!');
+  });
+
+  it('logs redis client errors.', async(): Promise<void> => {
+    const logger = { error: jest.fn() };
+    const mock = jest.spyOn(LogUtil, 'getLoggerFor');
+    mock.mockReturnValueOnce(logger as any);
+    locker = new RedisResourceLocker([ '6379' ]);
+    expect(logger.error).toHaveBeenCalledTimes(0);
+    redlock.emit('clientError', 'problem!');
+    expect(logger.error).toHaveBeenCalledTimes(1);
+    expect(logger.error).toHaveBeenLastCalledWith('Redis/Redlock error: problem!');
+  });
+
+  describe('createRedisClients', (): void => {
+    it('should create and return the right amount of redisClients.', async(): Promise<void> => {
+      // Reset calls done in `beforeEach`
+      jest.clearAllMocks();
+      const clientStrings = [ '6379', '127.0.0.1:6378' ];
+      locker = new RedisResourceLocker(clientStrings);
+      expect(createClient).toHaveBeenCalledTimes(2);
+      expect(createClient).toHaveBeenCalledWith(6379, undefined);
+      expect(createClient).toHaveBeenCalledWith(6378, '127.0.0.1');
+    });
+
+    it('errors when invalid string is passed.', async(): Promise<void> => {
+      // Reset calls done in `beforeEach`
+      jest.clearAllMocks();
+      const clientStrings = [ 'noHostOrPort' ];
+      expect((): any => new RedisResourceLocker(clientStrings))
+        .toThrow('Invalid data provided to create a Redis client: noHostOrPort');
+      expect(createClient).toHaveBeenCalledTimes(0);
+    });
+  });
+
+  describe('quit()', (): void => {
+    it('should clear all locks and intervals when quit() is called.', async(): Promise<void> => {
+      await locker.acquire(identifier);
+      await locker.quit();
+      expect(redlock.quit).toHaveBeenCalledTimes(1);
+
+      // This works since the Redlock is simply a mock and quit should have cleared the lockMap
+      await expect(locker.acquire(identifier)).resolves.toBeUndefined();
+    });
+  });
+});