Update Version Checks (#18)

This PR implements a required change for Trtl Anti-Entropy: Update
checks if the version being updated is still later than the local
version. I've also added a WithForce() option that bypasses checks if
the user wants to force put an object to the database, overriding the
namespace and version checks.

honu.go

@@ -138,11 +138,32 @@ func (db *DB) Update(obj *pb.Object, options ...opts.Option) (err error) {
 		return err
 	}
 
-	// Check the namespace and that it matches the object
-	if cfg.Namespace == opts.NamespaceDefault {
-		cfg.Namespace = obj.Namespace
-	} else if cfg.Namespace != obj.Namespace {
-		return errors.New("options namespace does not match object namespace")
+	if !cfg.Force {
+		// Check the namespace and that it matches the object
+		if cfg.Namespace == opts.NamespaceDefault {
+			cfg.Namespace = obj.Namespace
+		} else if cfg.Namespace != obj.Namespace {
+			return errors.New("options namespace does not match object namespace")
+		}
+
+		// Check that the version is later than the version being written to disk
+		var (
+			prevData []byte
+			prev     = new(pb.Object)
+		)
+		if prevData, err = tx.Get(obj.Key, cfg); err != nil {
+			if !errors.Is(err, engine.ErrNotFound) {
+				return fmt.Errorf("could not check previous version: %v", err)
+			}
+		} else {
+			if err = proto.Unmarshal(prevData, prev); err != nil {
+				return fmt.Errorf("could not unmarshal previous version: %v", err)
+			}
+		}
+
+		if !obj.Version.IsLater(prev.Version) {
+			return fmt.Errorf("cannot update object, it is not a later version then the current object")
+		}
 	}
 
 	// Put the version directly to disk

honu_test.go

@@ -126,11 +126,29 @@ func TestLevelDBInteractions(t *testing.T) {
 		require.Equal(t, "here", obj.Version.Region)
 
 		// Update with same namespace option should not error.
+		obj.Version.Version = 43
 		require.NoError(t, db.Update(obj, options.WithNamespace(namespace)))
 
 		// Update with wrong namespace should error
 		require.Error(t, db.Update(obj, options.WithNamespace("this is not the right thing")))
 
+		// Update with wrong namespace but with force should not error.
+		require.NoError(t, db.Update(obj, options.WithNamespace("trashcan"), options.WithForce()))
+
+		// Update with the same version should error.
+		require.Error(t, db.Update(obj, options.WithNamespace(namespace)))
+
+		// Update with an earlier version should error
+		obj.Version.Version = 7
+		require.Error(t, db.Update(obj, options.WithNamespace(namespace)))
+
+		// Update with an earlier version with force should not error.
+		require.NoError(t, db.Update(obj, options.WithNamespace(namespace), options.WithForce()))
+
+		// Update an object that does not exist should not error.
+		obj.Key = []byte("secretobjinvisible")
+		require.NoError(t, db.Update(obj, options.WithNamespace(namespace)))
+
 		// TODO: figure out what to do with this testcase.
 		// Iter currently grabs the namespace by splitting
 		// on :: and grabbing the first string, so it only

options/options.go

@@ -28,6 +28,7 @@ type Options struct {
 	LevelDBWrite *ldb.WriteOptions
 	PebbleWrite  *pebble.WriteOptions
 	Namespace    string
+	Force        bool
 }
 
 // Defines the signature of functions accepted as parameters by Honu methods.
@@ -44,6 +45,14 @@ func WithNamespace(namespace string) Option {
 	}
 }
 
+// WithForce prevents validation checks from returning an error during accesses.
+func WithForce() Option {
+	return func(cfg *Options) error {
+		cfg.Force = true
+		return nil
+	}
+}
+
 //Closure returning a function that adds the leveldbRead
 //parameter to an Options struct's LeveldbRead field.
 func WithLevelDBRead(opts *ldb.ReadOptions) Option {