Persisters should error if you try to save something which is gone now, but used to be persisted.

lib/valkyrie/persistence/composite_persister.rb

@@ -31,6 +31,8 @@ def save(resource:)
       # Don't pass opt lock tokens to other persisters
       internal_resource = cached_resource.dup
       internal_resource.clear_optimistic_lock_token!
+      # Reset new_record to avoid not-saved checks.
+      internal_resource.new_record = resource.new_record
       rest.inject(internal_resource) { |m, persister| persister.save(resource: m) }
       # return the one with the desired opt lock token
       cached_resource

lib/valkyrie/persistence/delete_tracking_buffer.rb

@@ -24,6 +24,12 @@ def delete(resource:)
         @deletes << resource
         super
       end
+
+      # A buffered memory persister can save anything, it doesn't have to hold
+      # a consistent internal state.
+      def valid_for_save?(_)
+        true
+      end
     end
   end
 end

lib/valkyrie/persistence/fedora/persister.rb

@@ -35,6 +35,8 @@ def save(resource:)
       alternate_resources ? save_reference_to_resource(persisted_resource, alternate_resources) : persisted_resource
     rescue Ldp::PreconditionFailed
       raise Valkyrie::Persistence::StaleObjectError, "The object #{internal_resource.id} has been updated by another process."
+    rescue Ldp::Gone
+      raise Valkyrie::Persistence::ObjectNotFoundError, "The object #{resource.id} is previously persisted but not found at save time."
     end
 
     # (see Valkyrie::Persistence::Memory::Persister#save_all)

lib/valkyrie/persistence/memory/persister.rb

@@ -5,7 +5,7 @@ module Valkyrie::Persistence::Memory
   # @note Documentation for persisters in general is maintained here.
   class Persister
     attr_reader :adapter
-    delegate :cache, to: :adapter
+    delegate :cache, :query_service, to: :adapter
 
     # @param adapter [Valkyrie::Persistence::Memory::MetadataAdapter] The memory adapter which
     #   holds the cache for this persister.
@@ -21,6 +21,7 @@ def initialize(adapter)
     # @raise [Valkyrie::Persistence::StaleObjectError]
     def save(resource:)
       raise Valkyrie::Persistence::StaleObjectError, "The object #{resource.id} has been updated by another process." unless valid_lock?(resource)
+      raise Valkyrie::Persistence::ObjectNotFoundError, "The object #{resource.id} is previously persisted but not found at save time." unless valid_for_save?(resource)
 
       # duplicate the resource so we are not creating side effects on the caller's resource
       internal_resource = resource.dup
@@ -34,6 +35,15 @@ def save(resource:)
       cache[internal_resource.id] = internal_resource
     end
 
+    # return true if resource is
+    # persisted and found
+    # or
+    # not persisted
+    def valid_for_save?(resource)
+      return true unless resource.persisted? # a new resource
+      query_service.find_by(id: resource.id).present? # a persisted resource must be found
+    end
+
     # Save a batch of resources.
     # @param resources [Array<Valkyrie::Resource>] List of resources to save.
     # @return [Array<Valkyrie::Resource>] List of resources with an `#id` value

lib/valkyrie/persistence/postgres/persister.rb

@@ -20,6 +20,7 @@ def initialize(adapter:)
     def save(resource:)
       orm_object = resource_factory.from_resource(resource: resource)
       orm_object.transaction do
+        raise Valkyrie::Persistence::ObjectNotFoundError, "The object #{resource.id} is previously persisted but not found at save time." if resource.persisted? && !orm_object.persisted?
         orm_object.save!
         if resource.id && resource.id.to_s != orm_object.id
           raise Valkyrie::Persistence::UnsupportedDatatype,

lib/valkyrie/persistence/solr/persister.rb

@@ -6,7 +6,7 @@ module Valkyrie::Persistence::Solr
   # Most methods are delegated to {Valkyrie::Persistence::Solr::Repository}
   class Persister
     attr_reader :adapter
-    delegate :connection, :resource_factory, :write_only?, to: :adapter
+    delegate :connection, :query_service, :resource_factory, :write_only?, to: :adapter
 
     # @param adapter [Valkyrie::Persistence::Solr::MetadataAdapter] The adapter with the
     #   configured solr connection.
@@ -17,13 +17,19 @@ def initialize(adapter:)
     # (see Valkyrie::Persistence::Memory::Persister#save)
     # @return [Boolean] If write_only, whether saving succeeded.
     def save(resource:)
+      raise Valkyrie::Persistence::ObjectNotFoundError, "The object #{resource.id} is previously persisted but not found at save time." unless write_only? || valid_for_save?(resource)
       if write_only?
         repository([resource]).persist
       else
         repository([resource]).persist.first
       end
     end
 
+    def valid_for_save?(resource)
+      return true unless resource.persisted? # a new resource
+      query_service.find_by(id: resource.id).present? # a persisted resource must be found
+    end
+
     # (see Valkyrie::Persistence::Memory::Persister#save_all)
     # @return [Boolean] If write_only, whether saving succeeded.
     def save_all(resources:)

lib/valkyrie/specs/shared_specs/persister.rb

@@ -50,6 +50,18 @@ class CustomResource < Valkyrie::Resource
     expect(reloaded.nested_resource.first.title).to eq ["Nested"]
   end
 
+  context "when a persisted resource is not in the database" do
+    it "throws an ObjectNotFoundError" do
+      expect(resource).not_to be_persisted
+      saved = persister.save(resource: resource)
+
+      expect(saved).to be_persisted
+      persister.delete(resource: saved)
+
+      expect { persister.save(resource: saved) }.to raise_error(Valkyrie::Persistence::ObjectNotFoundError)
+    end
+  end
+
   it "can persist single values" do
     resource.single_value = "A single value"
 

spec/valkyrie/persistence/buffered_persister_spec.rb

@@ -30,5 +30,15 @@ class Resource < Valkyrie::Resource
       end
       expect(buffer.query_service.find_all.length).to eq 1
     end
+    it "can handle saving an object which was previously saved" do
+      buffer = nil
+      resource = Resource.new
+      output = adapter.persister.save(resource: resource)
+      persister.with_buffer do |inner_persister, memory_buffer|
+        inner_persister.save(resource: output)
+        buffer = memory_buffer
+      end
+      expect(buffer.query_service.find_all.length).to eq 1
+    end
   end
 end