Rework Mutex

spec/client/watches/ApiSubject.spec.ts

@@ -795,4 +795,108 @@ describe('ApiSubject', function (): void {
 			).toBe(true)
 		})
 	}) // #canUpdate()
+
+	describe('concurrent operations', function (): void {
+		it('does not open the watch twice for two concurrent add calls', async function (): Promise<void> {
+			mockOpen()
+			asMock(apis.add).mockResolvedValue([secondDict])
+
+			// Both started concurrently before either has resolved. Only the
+			// first should open; the second must call addIds once the first
+			// releases the lock.
+			await Promise.all([subject.add(['foo']), subject.add(['bar'])])
+
+			expect(apis.open).toHaveBeenCalledTimes(1)
+			expect(apis.add).toHaveBeenCalledTimes(1)
+			expect(apis.add).toHaveBeenCalledWith('watchId', ['bar'])
+		})
+
+		it('processes concurrent add and remove in strict sequential order', async function (): Promise<void> {
+			mockOpen()
+			await subject.add(['foo'])
+
+			const callOrder: string[] = []
+
+			asMock(apis.add).mockImplementation(
+				async (): Promise<(typeof secondDict)[]> => {
+					callOrder.push('add')
+					return [secondDict]
+				}
+			)
+			asMock(apis.remove).mockImplementation(async (): Promise<void> => {
+				callOrder.push('remove')
+			})
+
+			// add and remove started without awaiting — must not interleave.
+			await Promise.all([subject.add(['bar']), subject.remove(['foo'])])
+
+			expect(callOrder).toEqual(['add', 'remove'])
+		})
+
+		it('releases the mutex when the network open call throws', async function (): Promise<void> {
+			const openResponse = {
+				id: firstGrid.meta.get<HStr>('watchId')?.value,
+				records: firstGrid.getRows(),
+			}
+
+			asMock(apis.open)
+				.mockRejectedValueOnce(new Error('network error'))
+				.mockResolvedValue(openResponse)
+
+			await expect(subject.add(['foo'])).rejects.toThrow('network error')
+
+			// The mutex must be released — a subsequent call should complete
+			// without hanging.
+			await subject.add(['foo'])
+			expect(apis.open).toHaveBeenCalledTimes(2)
+		})
+
+		it('releases the mutex when the network add call throws', async function (): Promise<void> {
+			mockOpen()
+			await subject.add(['foo'])
+
+			asMock(apis.add)
+				.mockRejectedValueOnce(new Error('network error'))
+				.mockResolvedValue([secondDict])
+
+			await expect(subject.add(['bar'])).rejects.toThrow('network error')
+
+			// The mutex must be released — a subsequent call should complete
+			// without hanging.
+			await subject.add(['bar'])
+			expect(apis.add).toHaveBeenCalledTimes(2)
+		})
+
+		it('releases the mutex when the network remove call throws', async function (): Promise<void> {
+			mockOpen()
+			await subject.add(['foo'])
+
+			asMock(apis.remove)
+				.mockRejectedValueOnce(new Error('network error'))
+				.mockResolvedValue(undefined)
+
+			await expect(subject.remove(['foo'])).rejects.toThrow(
+				'network error'
+			)
+
+			// Subsequent operations should still be able to run.
+			asMock(apis.add).mockResolvedValue([secondDict])
+			await subject.add(['bar'])
+			expect(apis.add).toHaveBeenCalledWith('watchId', ['bar'])
+		})
+
+		it('only adds ids to the server that are not already being watched', async function (): Promise<void> {
+			mockOpen()
+			asMock(apis.add).mockResolvedValue([secondDict])
+
+			// 'foo' is opened by the first call. The second call should only
+			// send 'bar' to the server, not 'foo' again.
+			await Promise.all([
+				subject.add(['foo']),
+				subject.add(['foo', 'bar']),
+			])
+
+			expect(apis.add).toHaveBeenCalledWith('watchId', ['bar'])
+		})
+	}) // concurrent operations
 })

spec/client/watches/BatchSubject.spec.ts

@@ -2,9 +2,12 @@
  * Copyright (c) 2021, J2 Innovations. All Rights Reserved
  */
 
-import { HGrid } from 'haystack-core'
+import { HGrid, HRef } from 'haystack-core'
 import { BatchSubject } from '../../../src/client/watches/BatchSubject'
-import { Subject } from '../../../src/client/watches/Subject'
+import {
+	Subject,
+	SubjectChangedEventHandler,
+} from '../../../src/client/watches/Subject'
 
 describe('BatchSubject', function (): void {
 	let subject: Subject
@@ -15,18 +18,73 @@ describe('BatchSubject', function (): void {
 		ops = []
 
 		subject = {
+			display: 'inner-display',
+			pollRate: 30,
 			add: jest.fn().mockImplementation(async (ids: string[]) => {
 				ops.push('add:' + ids.join(','))
 			}),
 			remove: jest.fn().mockImplementation(async (ids: string[]) => {
 				ops.push('remove:' + ids.join(','))
 			}),
-			update: jest.fn(),
+			refresh: jest.fn().mockResolvedValue(undefined),
+			update: jest.fn().mockResolvedValue(undefined),
+			on: jest.fn(),
+			off: jest.fn(),
+			get: jest.fn(),
+			inspect: jest.fn(),
 		} as unknown as Subject
 
 		batch = new BatchSubject(subject)
 	})
 
+	describe('#display', function (): void {
+		it('delegates to the inner subject', function (): void {
+			expect(batch.display).toBe('inner-display')
+		})
+	}) // #display
+
+	describe('#pollRate', function (): void {
+		it('delegates the getter to the inner subject', function (): void {
+			expect(batch.pollRate).toBe(30)
+		})
+
+		it('delegates the setter to the inner subject', function (): void {
+			batch.pollRate = 60
+			expect(subject.pollRate).toBe(60)
+		})
+	}) // #pollRate
+
+	describe('#refresh()', function (): void {
+		it('delegates to the inner subject', async function (): Promise<void> {
+			await batch.refresh()
+			expect(subject.refresh).toHaveBeenCalled()
+		})
+	}) // #refresh()
+
+	describe('#on()', function (): void {
+		it('delegates to the inner subject', function (): void {
+			const cb: SubjectChangedEventHandler = jest.fn()
+			batch.on(cb)
+			expect(subject.on).toHaveBeenCalledWith(cb)
+		})
+	}) // #on()
+
+	describe('#off()', function (): void {
+		it('delegates to the inner subject', function (): void {
+			const cb: SubjectChangedEventHandler = jest.fn()
+			batch.off(cb)
+			expect(subject.off).toHaveBeenCalledWith(cb)
+		})
+	}) // #off()
+
+	describe('#get()', function (): void {
+		it('delegates to the inner subject', function (): void {
+			const ref = HRef.make('test')
+			batch.get(ref)
+			expect(subject.get).toHaveBeenCalledWith(ref)
+		})
+	}) // #get()
+
 	describe('#add()', function (): void {
 		it("invokes inner subject's add", async function (): Promise<void> {
 			await batch.add(['a'])
@@ -65,6 +123,31 @@ describe('BatchSubject', function (): void {
 
 			expect(order).toBe('abcdefgh')
 		})
+
+		it('propagates inner subject error to the caller', async function (): Promise<void> {
+			const error = new Error('inner add failed')
+			;(subject.add as jest.Mock).mockRejectedValue(error)
+
+			await expect(batch.add(['a'])).rejects.toThrow('inner add failed')
+		})
+
+		it('propagates inner subject error to all coalesced callers', async function (): Promise<void> {
+			const error = new Error('inner add failed')
+			;(subject.add as jest.Mock).mockRejectedValue(error)
+
+			// All three queue behind the same batch window and coalesce into one inner call.
+			const results = await Promise.allSettled([
+				batch.add(['a']),
+				batch.add(['b']),
+				batch.add(['c']),
+			])
+
+			expect(results[0].status).toBe('rejected')
+			expect(results[1].status).toBe('rejected')
+			expect(results[2].status).toBe('rejected')
+			// The inner add should only have been invoked once for the batch.
+			expect(subject.add).toHaveBeenCalledTimes(1)
+		})
 	}) // #add()
 
 	describe('#remove()', function (): void {
@@ -108,8 +191,37 @@ describe('BatchSubject', function (): void {
 
 			expect(order).toBe('abcdefghi')
 		})
+
+		it('propagates inner subject error to the caller', async function (): Promise<void> {
+			const error = new Error('inner remove failed')
+			;(subject.remove as jest.Mock).mockRejectedValue(error)
+
+			await expect(batch.remove(['a'])).rejects.toThrow(
+				'inner remove failed'
+			)
+		})
 	}) // #remove()
 
+	describe('error handling', function (): void {
+		it('continues processing subsequent ops after a failed op', async function (): Promise<void> {
+			;(subject.add as jest.Mock).mockRejectedValueOnce(
+				new Error('add failed')
+			)
+			;(subject.remove as jest.Mock).mockResolvedValue(undefined)
+
+			// add and remove are different ops so they are not coalesced — they
+			// sit in the queue as separate entries and are processed in sequence.
+			const results = await Promise.allSettled([
+				batch.add(['a']),
+				batch.remove(['b']),
+			])
+
+			expect(results[0].status).toBe('rejected')
+			expect(results[1].status).toBe('fulfilled')
+			expect(subject.remove).toHaveBeenCalledWith(['b'])
+		})
+	}) // error handling
+
 	describe('#update()', () => {
 		it('calls inner subject update', async () => {
 			const grid = new HGrid()