Ack failed messages, improve concurrency, and make thread safe