Message deduplication

[snichme]

WHAT is this pull request doing?

Implementation for message deduplication on exchanges.
It works for any exchange type and is enabled by setting the argument x-message-deduplication = true on any exchange.

Features

• Deduplication on Exchanges
• Deduplication cache storage in memory
• TTL for each message
• Default TTL
• Deduplication on queues

Fixes #833

HOW can this pull request be tested?

Specs.

[snichme]

TODO: Expose the dedup_count as prometheus metric as well.

[snichme]

Documentation for the website: https://gist.github.com/snichme/40a33df7ea9a02c9a7d77a5c81d1f52d

[viktorerlingsson]

Should this file really be called dedup_ext.cr and be in the exchange dir? I'm thinking it doesn't seem to have much in common with other exchange types, and doesn't inherit Exchange. So maybe this should be called something like deduplication.cr instead and be in another dir?

[snichme]

Thats why I called it _ext for extension. There are mixed files in different places (like in queue/ there are are queues, stores and events) so that is why I did like this.

It's also used for both queues and exchanges so have it somewhere else makes sense. directly under /src/lavinmq maybe?

[viktorerlingsson]

Thats why I called it _ext for extension.

Okay 👍

There are mixed files in different places (like in queue/ there are are queues, stores and events) so that is why I did like this.

Hmm, yeah, you're right. Maybe we should give our overall file structure some more thought! 🙂

It's also used for both queues and exchanges so have it somewhere else makes sense. directly under /src/lavinmq maybe?

Yeah, I think directly under /src/lavinmq makes more sense! And if we add more "extensions" in the future we could move those to a separate extensions dir?

[carlhoerberg]

isn't the purpose of the TTL is to keep the size of the cache reasonable small? So I think you should have a fiber that checks the TTL of each key and evict the ones that has expired. Not check the TTL in the contains? method.

monotonic is a better alternative when you only are interested in the timespan, not the exact point of time.

Suggested change

	class MemoryCache(T) < Cache(T)
	def initialize(@size : UInt32)
	@store = Hash(T, Int64?).new
	end
	class MemoryCache(T) < Cache(T)
	def initialize(@size : UInt32)
	@store = Hash(T, Int64?).new
	spawn(name: "memorycache expiry") do
	loop do
	sleep 60.seconds
	@store.reject! { |_, ttd| tdd && RoughTime.monotonic > ttd }
	end
	end

[carlhoerberg]

thinking more about it, maybe i'm wrong about the purpose of the TTL here.

[carlhoerberg]

Suggested change

	return false unless @store.has_key?(key)
	ttd = @store[key]
	return true unless ttd
	return true if ttd > RoughTime.unix_ms
	@store.delete(key)
	false
	@store.has_key?(key)

[carlhoerberg]

Suggested change

	@store[key] = ttl ? RoughTime.unix_ms + ttl : nil
	@store[key] = ttl ? RoughTime.monotonic + ttl : nil

[carlhoerberg]

Suggested change

	@store = Hash(T, Int64?).new
	@store = Hash(T, Int64?).new(initial_size: @size)

[spuun]

I think the cache needs to be thread-safe?