diff --git a/lib/mongoid/association/many.rb b/lib/mongoid/association/many.rb
index 82f53a8e26..19b0eba092 100644
--- a/lib/mongoid/association/many.rb
+++ b/lib/mongoid/association/many.rb
@@ -178,6 +178,19 @@ def unscoped
         criteria.unscoped
       end
 
+      # For compatibility with Rails' caching. Returns a string based on the
+      # given timestamp, and includes the number of records in the relation
+      # in the version.
+      #
+      # @param [ String | Symbol ] timestamp_column the timestamp column to
+      #   use when constructing the key.
+      #
+      # @return [ String ] the cache version string
+      def cache_version(timestamp_column = :updated_at)
+        @cache_version ||= {}
+        @cache_version[timestamp_column] ||= compute_cache_version(timestamp_column)
+      end
+
       private
 
       def _session
@@ -198,6 +211,54 @@ def find_or(method, attrs = {}, type = nil, &block)
         attrs[klass.discriminator_key] = type.discriminator_value if type
         where(attrs).first || send(method, attrs, type, &block)
       end
+
+      # Computes the cache version for the relation using the given
+      # timestamp colum; see `#cache_version`.
+      #
+      # @param [ String | Symbol ] timestamp_column the timestamp column to
+      #   use when constructing the key.
+      #
+      # @return [ String ] the cache version string
+      def compute_cache_version(timestamp_column)
+        timestamp_column = timestamp_column.to_s
+
+        loaded = _target.respond_to?(:_loaded?) ?
+                    _target._loaded? :   # has_many
+                    true                 # embeds_many
+
+        size, timestamp = loaded ?
+          analyze_loaded_target(timestamp_column) :
+          analyze_unloaded_target(timestamp_column)
+
+        if timestamp
+          "#{size}-#{timestamp.utc.to_formatted_s(klass.cache_timestamp_format)}"
+        else
+          size.to_s
+        end
+      end
+
+      # Return a 2-tuple of the number of elements in the relation, and the
+      # largest timestamp value.
+      def analyze_loaded_target(timestamp_column)
+        newest = _target.select { |elem| elem.respond_to?(timestamp_column) }
+                        .max { |a, b| a[timestamp_column] <=> b[timestamp_column] }
+        [ _target.length, newest ? newest[timestamp_column] : nil ]
+      end
+
+      # Returns a 2-tuple of the number of elements in the relation, and the
+      # largest timestamp value. This will query the database to perform a
+      # $sum and a $max.
+      def analyze_unloaded_target(timestamp_column)
+        pipeline = criteria
+          .group(_id: nil,
+                 count: { '$sum' => 1 },
+                 latest: { '$max' => "$#{timestamp_column}" })
+          .pipeline
+
+        result = klass.collection.aggregate(pipeline).to_a.first
+
+        result ? [ result["count"], result["latest"] ] : [ 0 ]
+      end
     end
   end
 end
diff --git a/lib/mongoid/cacheable.rb b/lib/mongoid/cacheable.rb
index dae0e47317..8b271f51fe 100644
--- a/lib/mongoid/cacheable.rb
+++ b/lib/mongoid/cacheable.rb
@@ -15,20 +15,39 @@ module Cacheable
     # Print out the cache key. This will append different values on the
     # plural model name.
     #
-    # If new_record?     - will append /new
-    # If not             - will append /id-updated_at.to_formatted_s(cache_timestamp_format)
-    # Without updated_at - will append /id
+    # If new_record? - will append /new
+    # Non-nil cache_version? - append /id
+    # Non-nil updated_at - append /id-updated_at.to_formatted_s(cache_timestamp_format)
+    # Otherwise - append /id
     #
     # This is usually called inside a cache() block
     #
     # @example Returns the cache key
     #   document.cache_key
     #
-    # @return [ String ] the string with or without updated_at
+    # @return [ String ] the generated cache key
     def cache_key
       return "#{model_key}/new" if new_record?
+      return "#{model_key}/#{_id}" if cache_version
       return "#{model_key}/#{_id}-#{updated_at.utc.to_formatted_s(cache_timestamp_format)}" if try(:updated_at)
       "#{model_key}/#{_id}"
     end
+
+    # Return the cache version for this model. By default, it returns the updated_at
+    # field (if present) formatted as a string, or nil if the model has no
+    # updated_at field. Models with different needs may override this method to
+    # suit their desired behavior.
+    #
+    # @return [ String | nil ] the cache version value
+    #
+    # TODO: we can test this by using a MemoryStore, putting something in
+    # it, then updating the timestamp on the record and trying to read the
+    # value from the memory store. It shouldn't find it, because the version
+    # has changed.
+    def cache_version
+      if has_attribute?('updated_at') && updated_at.present?
+        updated_at.utc.to_formatted_s(cache_timestamp_format)
+      end
+    end
   end
 end
diff --git a/spec/integration/caching_spec.rb b/spec/integration/caching_spec.rb
new file mode 100644
index 0000000000..89842a664e
--- /dev/null
+++ b/spec/integration/caching_spec.rb
@@ -0,0 +1,68 @@
+# frozen_string_literal: true
+
+require 'spec_helper'
+
+describe 'caching integration tests' do
+  let(:store) { ActiveSupport::Cache::MemoryStore.new }
+
+  context 'without updated_at' do
+    let(:model1) { Person.create }
+    let(:model2) { Person.create }
+
+    before do
+      store.write(model1, 'model1')
+      store.write(model2, 'model2')
+    end
+
+    it 'uses a unique key' do
+      expect(store.read(model1)).to be == 'model1'
+      expect(store.read(model2)).to be == 'model2'
+    end
+
+    context 'when updating' do
+      before do
+        model1.update title: 'updated'
+        model2.update title: 'updated'
+      end
+
+      let(:reloaded_model1) { Person.find(model1.id) }
+      let(:reloaded_model2) { Person.find(model2.id) }
+
+      it 'still finds the models' do
+        expect(store.read(reloaded_model1)).to be == 'model1'
+        expect(store.read(reloaded_model2)).to be == 'model2'
+      end
+    end
+  end
+
+  context 'with updated_at' do
+    let(:model1) { Dokument.create }
+    let(:model2) { Dokument.create }
+
+    before do
+      store.write(model1, 'model1')
+      store.write(model2, 'model2')
+    end
+
+    it 'uses a unique key' do
+      expect(store.read(model1)).to be == 'model1'
+      expect(store.read(model2)).to be == 'model2'
+    end
+
+    context 'when updating' do
+      before do
+        model1.update title: 'updated'
+        model2.update title: 'updated'
+      end
+
+      let(:reloaded_model1) { Dokument.find(model1.id) }
+      let(:reloaded_model2) { Dokument.find(model2.id) }
+
+      it 'does not find the models' do
+        # because the update caused the cache_version to change
+        expect(store.read(reloaded_model1)).to be_nil
+        expect(store.read(reloaded_model2)).to be_nil
+      end
+    end
+  end
+end
diff --git a/spec/mongoid/association/embedded/embeds_many/proxy_spec.rb b/spec/mongoid/association/embedded/embeds_many/proxy_spec.rb
index 7ef43ae68c..28a8e9c9d7 100644
--- a/spec/mongoid/association/embedded/embeds_many/proxy_spec.rb
+++ b/spec/mongoid/association/embedded/embeds_many/proxy_spec.rb
@@ -4896,4 +4896,105 @@ class DNS::Record
       expect(user.orders.map(&:sku).sort).to eq([ 1, 2 ])
     end
   end
+
+  describe '#cache_version' do
+    context 'when the model does not have an updated_at column' do
+      let(:root_model) { Quiz.create! }
+      let(:root) { Quiz.find(root_model.id) }
+      let(:pages) { root.pages }
+
+      let(:prepopulated_root) do
+        root_model.pages << Page.new(content: 'Page #1')
+        root_model.pages << Page.new(content: 'Page #2')
+        Quiz.find(root_model.id)
+      end
+
+      shared_examples_for 'a cache_version generator' do
+        it 'produces a trivial cache_version' do
+          expect(pages.cache_version).to be == "#{pages.length}"
+        end
+      end
+
+      context 'when the relation is empty' do
+        it_behaves_like 'a cache_version generator'
+      end
+
+      context 'when the relation is not empty' do
+        let(:root) { prepopulated_root }
+
+        it_behaves_like 'a cache_version generator'
+      end
+    end
+
+    context 'when the model has an updated_at column' do
+      let(:root_model) { Book.create(title: 'Root') }
+      let(:root) { Book.find(root_model.id) }
+
+      let(:cover) { root_model.covers.first }
+      let(:covers) { root.covers }
+      let(:original_cache_version) { root.covers.cache_version }
+
+      let(:prepopulated_root) do
+        root_model.covers << Cover.new(title: 'Cover #1')
+        root_model.covers << Cover.new(title: 'Cover #2')
+        Book.find(root_model.id)
+      end
+
+      shared_examples_for 'a cache_version generator' do
+        it 'produces a consistent cache_version' do
+          expect(covers.cache_version).not_to be_nil
+          expect(covers.cache_version).to be == covers.cache_version
+        end
+      end
+
+      context 'when the relation is empty' do
+        it_behaves_like 'a cache_version generator'
+      end
+
+      context 'when the relation is not empty' do
+        let(:root) { prepopulated_root }
+        it_behaves_like 'a cache_version generator'
+      end
+
+      context 'when an element is updated' do
+        let(:updated_cache_version) do
+          cover.update title: 'modified'
+          cover.book.save!
+          cover.book.reload.covers.cache_version
+        end
+
+        let(:root) { prepopulated_root }
+
+        it 'changes the cache_version' do
+          expect(original_cache_version).not_to be == updated_cache_version
+        end
+      end
+
+      context 'when an element is added' do
+        let(:updated_cache_version) do
+          root.covers << Cover.new(title: 'Another Cover')
+          root.reload.covers.cache_version
+        end
+
+        let(:root) { prepopulated_root }
+
+        it 'changes the cache_version' do
+          expect(original_cache_version).not_to be == updated_cache_version
+        end
+      end
+
+      context 'when an element is removed' do
+        let(:updated_cache_version) do
+          cover.destroy
+          root.reload.covers.cache_version
+        end
+
+        let(:root) { prepopulated_root }
+
+        it 'changes the cache_version' do
+          expect(original_cache_version).not_to be == updated_cache_version
+        end
+      end
+    end
+  end
 end
diff --git a/spec/mongoid/association/referenced/has_many/proxy_spec.rb b/spec/mongoid/association/referenced/has_many/proxy_spec.rb
index 1e9f0d1b21..8ff9db60f5 100644
--- a/spec/mongoid/association/referenced/has_many/proxy_spec.rb
+++ b/spec/mongoid/association/referenced/has_many/proxy_spec.rb
@@ -1,4 +1,5 @@
 # frozen_string_literal: true
+# rubocop:todo all
 
 require 'spec_helper'
 
@@ -2530,13 +2531,11 @@ def with_transaction_via(model, &block)
     let(:post_one) { Post.create!(rating: 5) }
     let(:post_two) { Post.create!(rating: 10) }
 
-    # rubocop:disable Performance/CompareWithBlock
     let(:max) do
       person.posts.max do |a, b|
         a.rating <=> b.rating
       end
     end
-    # rubocop:enable Performance/CompareWithBlock
 
     before do
       person.posts.push(post_one, post_two)
@@ -2619,13 +2618,11 @@ def with_transaction_via(model, &block)
     let(:post_one) { Post.create!(rating: 5) }
     let(:post_two) { Post.create!(rating: 10) }
 
-    # rubocop:disable Performance/CompareWithBlock
     let(:min) do
       person.posts.min do |a, b|
         a.rating <=> b.rating
       end
     end
-    # rubocop:enable Performance/CompareWithBlock
 
     before do
       person.posts.push(post_one, post_two)
@@ -3230,4 +3227,137 @@ def with_transaction_via(model, &block)
       expect(agent.basic_ids).to eq [ basic.id ]
     end
   end
+
+  describe '#cache_version' do
+    context 'when the model does not have an updated_at column' do
+      let(:root_model) { Person.create! }
+      let(:root) { Person.find(root_model.id) }
+
+      let(:prepopulated_root) do
+        root_model.posts << Post.new(title: 'Post #1')
+        root_model.posts << Post.new(title: 'Post #2')
+        Person.find(root_model.id)
+      end
+
+      shared_examples_for 'a cache_version generator' do
+        it 'produces a trivial cache_version' do
+          expect(posts.cache_version).to be == "#{posts.length}"
+        end
+      end
+
+      context 'when the relation is already loaded' do
+        let(:posts) { root.posts.tap { |r| r.to_a } }
+
+        context 'when the relation is empty' do
+          it_behaves_like 'a cache_version generator'
+        end
+
+        context 'when the relation is not empty' do
+          let(:root) { prepopulated_root }
+
+          it_behaves_like 'a cache_version generator'
+        end
+      end
+
+      context 'when the relation is not already loaded' do
+        let(:posts) { root.posts }
+
+        context 'when the relation is empty' do
+          it_behaves_like 'a cache_version generator'
+        end
+
+        context 'when the relation is not empty' do
+          let(:root) { prepopulated_root }
+
+          it_behaves_like 'a cache_version generator'
+        end
+      end
+    end
+
+    context 'when the model has an updated_at column' do
+      let(:root_model) { WikiPage.create(title: 'Root') }
+      let(:root) { WikiPage.find(root_model.id) }
+
+      let(:child_page) { root_model.child_pages.first }
+      let(:original_cache_version) { root.child_pages.cache_version }
+
+      let(:prepopulated_root) do
+        root_model.child_pages << WikiPage.new(title: 'Child #1')
+        root_model.child_pages << WikiPage.new(title: 'Child #2')
+        WikiPage.find(root_model.id)
+      end
+
+      shared_examples_for 'a cache_version generator' do
+        it 'produces a consistent cache_version' do
+          expect(child_pages.cache_version).not_to be_nil
+          expect(child_pages.cache_version).to be == child_pages.cache_version
+        end
+      end
+
+      context 'when the relation is already loaded' do
+        let(:child_pages) { root.child_pages.tap { |r| r.to_a } }
+
+        context 'when the relation is empty' do
+          it_behaves_like 'a cache_version generator'
+        end
+
+        context 'when the relation is not empty' do
+          let(:root) { prepopulated_root }
+          it_behaves_like 'a cache_version generator'
+        end
+      end
+
+      context 'when the relation is not yet loaded' do
+        let(:child_pages) { root.child_pages }
+
+        context 'when the relation is empty' do
+          it_behaves_like 'a cache_version generator'
+        end
+
+        context 'when the relation is not empty' do
+          let(:root) { prepopulated_root }
+          it_behaves_like 'a cache_version generator'
+        end
+      end
+
+      context 'when an element is updated' do
+        let(:updated_cache_version) do
+          child_page.update description: 'modified'
+          root.reload.child_pages.cache_version
+        end
+
+        let(:root) { prepopulated_root }
+
+        it 'changes the cache_version' do
+          expect(original_cache_version).not_to be == updated_cache_version
+        end
+      end
+
+      context 'when an element is added' do
+        let(:updated_cache_version) do
+          root.child_pages << WikiPage.new(title: 'Another Child')
+          root.reload.child_pages.cache_version
+        end
+
+        let(:root) { prepopulated_root }
+
+        it 'changes the cache_version' do
+          expect(original_cache_version).not_to be == updated_cache_version
+        end
+      end
+
+      context 'when an element is removed' do
+        let(:updated_cache_version) do
+          child_page.destroy
+          root.reload.child_pages.cache_version
+        end
+
+        let(:root) { prepopulated_root }
+
+        it 'changes the cache_version' do
+          expect(original_cache_version).not_to be == updated_cache_version
+        end
+      end
+    end
+  end
 end
diff --git a/spec/mongoid/cacheable_spec.rb b/spec/mongoid/cacheable_spec.rb
index af5640ab06..b633496db9 100644
--- a/spec/mongoid/cacheable_spec.rb
+++ b/spec/mongoid/cacheable_spec.rb
@@ -41,35 +41,49 @@
         document.save!
       end
 
-      context "with updated_at" do
-
-        context "with the default cache_timestamp_format" do
-
-          let!(:updated_at) do
-            document.updated_at.utc.to_formatted_s(:nsec)
+      context 'with cache_version' do
+        context 'with updated_at' do
+          it 'does not include the timestamp in the key' do
+            expect(document.cache_key).to eq("dokuments/#{document.id}")
           end
+        end
+      end
 
-          it "has the id and updated_at key name" do
-            expect(document.cache_key).to eq("dokuments/#{document.id}-#{updated_at}")
-          end
+      context 'without cache_version' do
+        before do
+          document.define_singleton_method(:cache_version) { nil }
         end
 
-        context "with a different cache_timestamp_format" do
+        context "with updated_at" do
 
-          before do
-            Dokument.cache_timestamp_format = :number
-          end
+          context "with the default cache_timestamp_format" do
 
-          after do
-            Dokument.cache_timestamp_format = :nsec
-          end
+            let!(:updated_at) do
+              document.updated_at.utc.to_formatted_s(:nsec)
+            end
 
-          let!(:updated_at) do
-            document.updated_at.utc.to_formatted_s(:number)
+            it "has the id and updated_at key name" do
+              expect(document.cache_key).to eq("dokuments/#{document.id}-#{updated_at}")
+            end
           end
 
-          it "has the id and updated_at key name" do
-            expect(document.cache_key).to eq("dokuments/#{document.id}-#{updated_at}")
+          context "with a different cache_timestamp_format" do
+
+            before do
+              Dokument.cache_timestamp_format = :number
+            end
+
+            after do
+              Dokument.cache_timestamp_format = :nsec
+            end
+
+            let!(:updated_at) do
+              document.updated_at.utc.to_formatted_s(:number)
+            end
+
+            it "has the id and updated_at key name" do
+              expect(document.cache_key).to eq("dokuments/#{document.id}-#{updated_at}")
+            end
           end
         end
       end
@@ -112,4 +126,32 @@
       end
     end
   end
+
+  describe 'cache_version' do
+    context 'when model has no updated_at field' do
+      let(:model) { Artist.create! }
+
+      it 'has a nil cache_version' do
+        expect(model.cache_version).to be_nil
+      end
+    end
+
+    context 'when model has updated_at field' do
+      context 'when updated_at is nil' do
+        let(:model) { Dokument.create!.tap { |v| v.updated_at = nil } }
+
+        it 'has a nil cache_version' do
+          expect(model.cache_version).to be_nil
+        end
+      end
+
+      context 'when updated_at is present' do
+        let(:model) { Dokument.create! }
+
+        it 'has a non-nil cache_version' do
+          expect(model.cache_version).not_to be_nil
+        end
+      end
+    end
+  end
 end