Handle concurrency errors (see #31)

src/relay.js

@@ -51,7 +51,7 @@ export default class Relay extends EventEmitter {
     const num = batch.length
     debug(`submit: submitting ${num} item(s)`)
     return this._client.submit(batch)
-      .then(() => this._onSubmitted(num), (err) => this._onError(err))
+      .then(() => this._onSubmitted(num), (err) => this._onError(err, num))
       .then(() => this._scheduleSubmission())
   }
 
@@ -60,8 +60,19 @@ export default class Relay extends EventEmitter {
     this._queue.remove(num)
   }
 
-  _onError (err) {
+  _onError (err, num) {
     debug('onError', {error: err})
-    this.emit('error', err)
+    // Expected errors:
+    // - DataAlreadyAcceptedException
+    //   Message: "The given batch of log events has already been accepted."
+    //   Action: Assume the request got replayed and remove the batch.
+    // - InvalidSequenceTokenException
+    //   Message: "The given sequenceToken is invalid."
+    //   Action: Keep the items in the queue and retry next time.
+    if (err.code === 'DataAlreadyAcceptedException') {
+      this._queue.remove(num)
+    } else if (err.code !== 'InvalidSequenceTokenException') {
+      this.emit('error', err)
+    }
   }
 }

test/lib/client-mock.js

@@ -1,16 +1,19 @@
 export default class MockClient {
-  constructor (failures = 0) {
+  constructor (failures = []) {
     this._submitted = []
-    this._failures = failures
+    this._failures = failures.slice()
   }
 
   submit (batch) {
-    this._submitted = this._submitted.concat(batch)
-    if (this._failures > 0) {
-      this._failures--
-      return Promise.reject(new Error('FAIL'))
+    if (this._failures.length === 0) {
+      this._submitted = this._submitted.concat(batch)
+      return Promise.resolve()
+    } else {
+      const code = this._failures.shift()
+      const error = new Error(code)
+      error.code = code
+      return Promise.reject(error)
     }
-    return Promise.resolve()
   }
 
   get submitted () {

test/unit/relay.spec.js

@@ -55,16 +55,39 @@ describe('Relay', () => {
 
     it('emits an error event', async () => {
       const submissionInterval = 50
-      const failures = 3
-      const retries = 2
+      const failures = ['FAIL', 'FAIL', 'FAIL']
       const spy = sinon.spy()
       const client = new ClientMock(failures)
       const relay = new Relay(client, {submissionInterval})
       relay.on('error', spy)
       relay.start()
       relay.submit({})
-      await delay(submissionInterval * (failures + retries) * 1.1)
-      expect(spy.callCount).to.equal(failures)
+      await delay(submissionInterval * failures.length * 1.1)
+      expect(spy.callCount).to.equal(failures.length)
+    })
+
+    it('silently handles a DataAlreadyAcceptedException error', async () => {
+      const submissionInterval = 50
+      const failures = ['DataAlreadyAcceptedException']
+      const spy = sinon.spy()
+      const client = new ClientMock(failures)
+      const relay = new Relay(client, {submissionInterval})
+      relay.on('error', spy)
+      relay.start()
+      relay.submit({})
+      await delay(submissionInterval * failures.length * 1.1)
+      expect(spy.callCount).to.equal(0)
+    })
+
+    it('handles InvalidSequenceTokenException errors by retrying', async () => {
+      const submissionInterval = 50
+      const failures = ['InvalidSequenceTokenException', 'InvalidSequenceTokenException']
+      const client = new ClientMock(failures)
+      const relay = new Relay(client, {submissionInterval})
+      relay.start()
+      relay.submit({})
+      await delay(submissionInterval * (failures.length + 1) * 1.1)
+      expect(client.submitted.length).to.equal(1)
     })
   })
 })