canceling a deferred redis method causes replies to get out of sync

[dpkp]

Test code:

@defer.inlineCallbacks
def test_cancel(self):
    db = yield redis.Connection(REDIS_HOST, REDIS_PORT)

    prefix = 'txredisapi:cancel'

    # Simple Set + Get
    key = prefix + '1'
    value = 'first'

    # set should return 'OK'
    res = yield db.set(key, value)
    self.assertEquals('OK', res)

    # get should return the value
    val = yield db.get(key)
    self.assertEquals(val, value)

    # Cancel a method, replies get out of sync
    d = db.time()
    d.addErrback(lambda _: True)
    d.cancel()

    # Set should return 'OK'
    # but note that failure shows it actually returns the result of the cancelled method...
    key = prefix + '2'
    value = 'second'
    res = yield db.set(key, value)
    self.assertEquals('OK', res)

    # Get should return value
    # but note that it actually returns the result of the previous set method...
    val = yield db.get(key)
    self.assertEquals(val, value)

    yield db.disconnect()

The issue appears to be that txredisapi uses a DeferredQueue to manage replies. The redis method returns a deferred from the DeferredQueue, waiting on the next reply to be written by the protocol to the queue. But if the method is cancelled before it consumes its reply from the queue, twisted will remove the deferred from the DeferredQueue waiting list. The reply data, however, is still written to the queue when received and is delivered to the next waiting deferred instead of the deferred that was cancelled and removed from the list. Now all future method calls yield incorrect results.

See https://github.com/twisted/twisted/blob/trunk/twisted/internet/defer.py#L1493-L1505