CommunitySolidServer
diff --git a/src/storage/LockingResourceStore.ts b/src/storage/LockingResourceStore.ts
@@ -16,6 +16,7 @@ import type { ChangeMap, ResourceStore } from './ResourceStore';
  * Store that for every call acquires a lock before executing it on the requested resource,
  * and releases it afterwards.
  * In case the request returns a Representation the lock will only be released when the data stream is finished.
+ * Similarly, for write operations the lock will be maintained as long as the incoming data stream is being read.
  *
  * For auxiliary resources the lock will be applied to the subject resource.
  * The actual operation is still executed on the auxiliary resource.
@@ -60,8 +61,9 @@ export class LockingResourceStore implements AtomicResourceStore {
     representation: Representation,
     conditions?: Conditions,
   ): Promise<ChangeMap> {
-    return this.locks.withWriteLock(
+    return this.lockedRepresentationWrite(
       this.getLockIdentifier(container),
+      representation,
       async(): Promise<ChangeMap> => this.source.addResource(container, representation, conditions),
     );
   }
@@ -71,8 +73,9 @@ export class LockingResourceStore implements AtomicResourceStore {
     representation: Representation,
     conditions?: Conditions,
   ): Promise<ChangeMap> {
-    return this.locks.withWriteLock(
+    return this.lockedRepresentationWrite(
       this.getLockIdentifier(identifier),
+      representation,
       async(): Promise<ChangeMap> => this.source.setRepresentation(identifier, representation, conditions),
     );
   }
@@ -89,8 +92,9 @@ export class LockingResourceStore implements AtomicResourceStore {
     patch: Patch,
     conditions?: Conditions,
   ): Promise<ChangeMap> {
-    return this.locks.withWriteLock(
+    return this.lockedRepresentationWrite(
       this.getLockIdentifier(identifier),
+      patch,
       async(): Promise<ChangeMap> => this.source.modifyResource(identifier, patch, conditions),
     );
   }
@@ -143,6 +147,66 @@ export class LockingResourceStore implements AtomicResourceStore {
     });
   }
 
+  /**
+   * Acquires a write lock and executes the given function,
+   * adapting the incoming data stream to reset the timer every time data is read.
+   * The stream is adapted in place, as source stores do not expect
+   * to receive a different representation than the one that was passed in.
+   * Should the lock expire before the function has finished, the data stream will be destroyed.
+   *
+   * @param identifier - Identifier that should be locked.
+   * @param representation - Representation whose data will be consumed while the resource is locked.
+   * @param whileLocked - Function to be executed while the resource is locked.
+   */
+  protected async lockedRepresentationWrite(
+    identifier: ResourceIdentifier,
+    representation: Representation,
+    whileLocked: () => Promise<ChangeMap>,
+  ): Promise<ChangeMap> {
+    const { data } = representation;
+    const ownReadDescriptor = Object.getOwnPropertyDescriptor(data, 'read');
+    // This method is restored by identity and invoked with an explicit receiver below.
+    // eslint-disable-next-line @typescript-eslint/unbound-method
+    const originalRead = data.read;
+    let writeActive = false;
+    function restoreRead(): void {
+      if (ownReadDescriptor) {
+        Object.defineProperty(data, 'read', ownReadDescriptor);
+      } else {
+        Reflect.deleteProperty(data, 'read');
+      }
+      writeActive = false;
+    }
+    try {
+      return await this.locks.withWriteLock(identifier, async(maintainLock): Promise<ChangeMap> => {
+        writeActive = true;
+        // Reset the timeout timer every time data is read while the lock is active
+        Object.defineProperty(data, 'read', {
+          configurable: ownReadDescriptor?.configurable ?? true,
+          enumerable: ownReadDescriptor?.enumerable ?? false,
+          writable: ownReadDescriptor?.writable ?? true,
+          value(this: Readable, size?: number): unknown {
+            maintainLock();
+            return originalRead.call(data, size);
+          },
+        });
+        try {
+          return await whileLocked();
+        } finally {
+          restoreRead();
+        }
+      });
+    } catch (error: unknown) {
+      // Destroy the data stream in case the lock expired while the data was still being consumed,
+      // otherwise the source store could keep on writing data without the protection of the lock.
+      if (writeActive) {
+        restoreRead();
+        data.destroy(error as Error);
+      }
+      throw error;
+    }
+  }
+
   /**
    * Wraps a representation to make it reset the timeout timer every time data is read.
    *

diff --git a/test/integration/LockingResourceStore.test.ts b/test/integration/LockingResourceStore.test.ts
@@ -14,7 +14,7 @@ import type { ExpiringReadWriteLocker } from '../../src/util/locking/ExpiringRea
 import { MemoryResourceLocker } from '../../src/util/locking/MemoryResourceLocker';
 import type { ReadWriteLocker } from '../../src/util/locking/ReadWriteLocker';
 import { WrappedExpiringReadWriteLocker } from '../../src/util/locking/WrappedExpiringReadWriteLocker';
-import { guardedStreamFrom } from '../../src/util/StreamUtil';
+import { endOfStream, guardedStreamFrom } from '../../src/util/StreamUtil';
 import { PIM, RDF } from '../../src/util/Vocabularies';
 import { SimpleSuffixStrategy } from '../util/SimpleSuffixStrategy';
 import { flushPromises } from '../util/Util';
@@ -28,6 +28,7 @@ describe('A LockingResourceStore', (): void => {
   let expiringLocker: ExpiringReadWriteLocker;
   let source: ResourceStore;
   let getRepresentationSpy: jest.SpyInstance;
+  let setRepresentationSpy: jest.SpyInstance;
 
   beforeEach(async(): Promise<void> => {
     jest.clearAllMocks();
@@ -63,6 +64,12 @@ describe('A LockingResourceStore', (): void => {
 
     // Make sure something is in the store before we read from it in our tests.
     await source.setRepresentation({ path }, new BasicRepresentation([ 1, 2, 3 ], APPLICATION_OCTET_STREAM));
+
+    // Simulates a source store that consumes the incoming data before resolving
+    setRepresentationSpy = jest.spyOn(source, 'setRepresentation');
+    setRepresentationSpy.mockImplementation(
+      async(identifier, representation: Representation): Promise<any> => endOfStream(representation.data),
+    );
   });
 
   it('destroys the stream when nothing is read after 1000ms.', async(): Promise<void> => {
@@ -110,4 +117,66 @@ describe('A LockingResourceStore', (): void => {
     // Verify the lock was acquired and released at the right time
     expect(getRepresentationSpy).toHaveBeenCalledTimes(1);
   });
+
+  it('destroys the incoming stream when nothing is written after 1000ms.', async(): Promise<void> => {
+    const representation = new BasicRepresentation([ 1, 2, 3 ], APPLICATION_OCTET_STREAM);
+    const errorCallback = jest.fn();
+    representation.data.on('error', errorCallback);
+
+    // Catch the expected error so its rejection is handled before the timer fires
+    let error: unknown;
+    const promise = store.setRepresentation({ path }, representation).catch((err: unknown): void => {
+      error = err;
+    });
+    // Allow the lock to be acquired
+    await flushPromises();
+    expect(setRepresentationSpy).toHaveBeenCalledTimes(1);
+
+    // Wait 1000ms without writing
+    jest.advanceTimersByTime(1000);
+    await flushPromises();
+    expect(representation.data.destroyed).toBe(true);
+    await promise;
+    expect(error).toEqual(new InternalServerError(`Lock expired after 1000ms on ${path}`));
+
+    // Verify a timeout error was thrown
+    expect(errorCallback).toHaveBeenCalledTimes(1);
+    expect(errorCallback).toHaveBeenLastCalledWith(new InternalServerError(`Lock expired after 1000ms on ${path}`));
+  });
+
+  it('destroys the incoming stream when pauses between writes exceed 1000ms.', async(): Promise<void> => {
+    const representation = new BasicRepresentation([ 1, 2, 3 ], APPLICATION_OCTET_STREAM);
+    const errorCallback = jest.fn();
+    representation.data.on('error', errorCallback);
+
+    // Catch the expected error so its rejection is handled before the timer fires
+    let error: unknown;
+    const promise = store.setRepresentation({ path }, representation).catch((err: unknown): void => {
+      error = err;
+    });
+    // Allow the lock to be acquired
+    await flushPromises();
+    expect(setRepresentationSpy).toHaveBeenCalledTimes(1);
+
+    // Wait 750ms and read
+    jest.advanceTimersByTime(750);
+    expect(representation.data.destroyed).toBe(false);
+    representation.data.read();
+
+    // Wait 750ms and read
+    jest.advanceTimersByTime(750);
+    expect(representation.data.destroyed).toBe(false);
+    representation.data.read();
+
+    // Wait 1000ms and watch the stream be destroyed
+    jest.advanceTimersByTime(1000);
+    await flushPromises();
+    expect(representation.data.destroyed).toBe(true);
+    await promise;
+    expect(error).toEqual(new InternalServerError(`Lock expired after 1000ms on ${path}`));
+
+    // Verify a timeout error was thrown
+    expect(errorCallback).toHaveBeenCalledTimes(1);
+    expect(errorCallback).toHaveBeenLastCalledWith(new InternalServerError(`Lock expired after 1000ms on ${path}`));
+  });
 });
diff --git a/test/unit/storage/LockingResourceStore.test.ts b/test/unit/storage/LockingResourceStore.test.ts
@@ -18,7 +18,7 @@ function emptyFn(): void {
 describe('A LockingResourceStore', (): void => {
   const auxiliaryId = { path: 'http://test.com/foo.dummy' };
   const subjectId = { path: 'http://test.com/foo' };
-  const data = { data: 'data!' } as any;
+  let data: Representation;
   let store: LockingResourceStore;
   let locker: jest.Mocked<ExpiringReadWriteLocker>;
   let source: ResourceStore;
@@ -33,6 +33,8 @@ describe('A LockingResourceStore', (): void => {
       return input;
     }
 
+    data = { data: guardedStreamFrom([ 1, 2, 3 ]) } as any;
+
     const readable = guardedStreamFrom([ 1, 2, 3 ]);
     const destroy = readable.destroy.bind(readable);
     jest.spyOn(readable, 'destroy').mockImplementation((error): any => destroy.call(readable, error));
@@ -70,7 +72,12 @@ describe('A LockingResourceStore', (): void => {
       ): Promise<T> => {
         order.push('lock write');
         try {
-          return await whileLocked(emptyFn);
+          // Allows simulating a timeout event
+          const timeout = new Promise<never>((resolve, reject): any => timeoutTrigger.on('timeout', (): void => {
+            order.push('timeout');
+            reject(new Error('timeout'));
+          }));
+          return await Promise.race([ Promise.resolve(whileLocked(emptyFn)), timeout ]);
         } finally {
           order.push('unlock write');
         }
@@ -159,6 +166,83 @@ describe('A LockingResourceStore', (): void => {
     expect(order).toEqual([ 'lock write', 'modifyResource', 'unlock write' ]);
   });
 
+  it('resets the write lock expiration every time incoming data is read.', async(): Promise<void> => {
+    const originalRead = jest.spyOn(data.data, 'read');
+    const maintainLock = jest.fn();
+    locker.withWriteLock.mockImplementationOnce((async <T>(
+      identifier: ResourceIdentifier,
+      whileLocked: (maintain: () => void) => PromiseOrValue<T>,
+    ): Promise<T> => whileLocked(maintainLock)) satisfies ReadWriteLocker['withWriteLock'] as any);
+    jest.spyOn(source, 'setRepresentation').mockImplementation(
+      async(identifier: ResourceIdentifier, representation: Representation): Promise<any> => {
+        representation.data.read();
+        representation.data.read();
+        order.push('setRepresentation');
+      },
+    );
+
+    await store.setRepresentation(subjectId, data);
+    expect(locker.withWriteLock).toHaveBeenCalledTimes(1);
+    expect(source.setRepresentation).toHaveBeenCalledTimes(1);
+    expect(source.setRepresentation).toHaveBeenLastCalledWith(subjectId, data, undefined);
+    expect(maintainLock).toHaveBeenCalledTimes(2);
+
+    // The original read function is restored once the write is finished
+    expect(data.data.read).toBe(originalRead);
+    data.data.read();
+    expect(maintainLock).toHaveBeenCalledTimes(2);
+  });
+
+  it('destroys the incoming data stream if the write lock expires.', async(): Promise<void> => {
+    const originalRead = jest.spyOn(data.data, 'read');
+    const destroy = jest.spyOn(data.data, 'destroy');
+    jest.spyOn(source, 'setRepresentation').mockImplementation((): any => {
+      order.push('useless set');
+      // This will never resolve
+      return new Promise(emptyFn);
+    });
+
+    const prom = store.setRepresentation(subjectId, data);
+
+    timeoutTrigger.emit('timeout');
+
+    await expect(prom).rejects.toThrow('timeout');
+    expect(locker.withWriteLock).toHaveBeenCalledTimes(1);
+    expect(source.setRepresentation).toHaveBeenCalledTimes(1);
+    expect(destroy).toHaveBeenCalledTimes(1);
+    expect(destroy).toHaveBeenLastCalledWith(new Error('timeout'));
+    expect(data.data.read).toBe(originalRead);
+    expect(order).toEqual([ 'lock write', 'useless set', 'timeout', 'unlock write' ]);
+  });
+
+  it('does not destroy the incoming data stream if the write itself errors.', async(): Promise<void> => {
+    const destroy = jest.spyOn(data.data, 'destroy');
+    jest.spyOn(source, 'setRepresentation').mockImplementation((): any => {
+      order.push('bad set');
+      throw new Error('dummy');
+    });
+
+    await expect(store.setRepresentation(subjectId, data)).rejects.toThrow('dummy');
+    expect(locker.withWriteLock).toHaveBeenCalledTimes(1);
+    expect(source.setRepresentation).toHaveBeenCalledTimes(1);
+    expect(destroy).toHaveBeenCalledTimes(0);
+    expect(order).toEqual([ 'lock write', 'bad set', 'unlock write' ]);
+  });
+
+  it('does not destroy the incoming data stream if the write lock can not be acquired.', async(): Promise<void> => {
+    const destroy = jest.spyOn(data.data, 'destroy');
+    locker.withWriteLock.mockImplementationOnce(async(): Promise<never> => {
+      order.push('failed lock');
+      throw new Error('lock error');
+    });
+
+    await expect(store.setRepresentation(subjectId, data)).rejects.toThrow('lock error');
+    expect(locker.withWriteLock).toHaveBeenCalledTimes(1);
+    expect(source.setRepresentation).toHaveBeenCalledTimes(0);
+    expect(destroy).toHaveBeenCalledTimes(0);
+    expect(order).toEqual([ 'failed lock' ]);
+  });
+
   it('releases the lock if an error was thrown.', async(): Promise<void> => {
     source.getRepresentation = async(): Promise<any> => {
       order.push('bad get');