MM-21672: KVCompareAndSet improvements (#13858)

* allow ExpireInSeconds < 0

Allow `ExpireInSeconds < 0` for use with `KVSetWithOptions`. While this has no practical use in reality, it's much easier to thoroughly unit tests the underlying functionality if we can match the semantics of CompareAndSet. Strictly speaking, this is a breaking change, but not relative to the advertised semantics. Anyway, it's also not entirely unreasonable to treat a negative `ExpireInSeconds` as having already expired vs. marking it as never expired.

* updated tests, to break apart

* honour expiry in CompareAndSet

* honour expiry in CompareAndDelete

* honour expiry in List

* fail unique constraint exception for SaveOrUpdate

A unique constraint error on a `SaveOrUpdate` should not be ignored: we did not save or update the requested value, as someone else managed to write the record first.

Note this is handled differently in `CompareAndSet`, where we correctly swallow the error and return `false` to indicate we did not successfully save the value.

* unexport DEFAULT_PLUGIN_KEY_FETCH_LIMIT

* s/InternalServerError/BadRequest/ for failed SaveOrUpdate

Co-authored-by: mattermod <mattermod@users.noreply.github.com>

model/plugin_kvset_options.go

@@ -32,7 +32,7 @@ func (opt *PluginKVSetOptions) IsValid() *AppError {
 // NewPluginKeyValueFromOptions return a PluginKeyValue given a pluginID, a KV pair and options.
 func NewPluginKeyValueFromOptions(pluginId, key string, value []byte, opt PluginKVSetOptions) (*PluginKeyValue, *AppError) {
 	expireAt := int64(0)
-	if opt.ExpireInSeconds > 0 {
+	if opt.ExpireInSeconds != 0 {
 		expireAt = GetMillis() + (opt.ExpireInSeconds * 1000)
 	}
 

store/sqlstore/plugin_store.go

@@ -14,7 +14,7 @@ import (
 )
 
 const (
-	DEFAULT_PLUGIN_KEY_FETCH_LIMIT = 10
+	defaultPluginKeyFetchLimit = 10
 )
 
 type SqlPluginStore struct {
@@ -60,12 +60,7 @@ func (ps SqlPluginStore) SaveOrUpdate(kv *model.PluginKeyValue) (*model.PluginKe
 		} else if rowsAffected == 0 {
 			// No rows were affected by the update, so let's try an insert
 			if err := ps.GetMaster().Insert(kv); err != nil {
-				// If the error is from unique constraints violation, it's the result of a
+				return nil, model.NewAppError("SqlPluginStore.SaveOrUpdate", "store.sql_plugin_store.save.app_error", nil, err.Error(), http.StatusBadRequest)
-				// valid race and we can report success. Otherwise we have a real error and
-				// need to return it
-				if !IsUniqueConstraintError(err, []string{"PRIMARY", "PluginId", "Key", "PKey", "pkey"}) {
-					return nil, model.NewAppError("SqlPluginStore.SaveOrUpdate", "store.sql_plugin_store.save.app_error", nil, err.Error(), http.StatusInternalServerError)
-				}
 			}
 		}
 	} else if ps.DriverName() == model.DATABASE_DRIVER_MYSQL {
@@ -88,6 +83,16 @@ func (ps SqlPluginStore) CompareAndSet(kv *model.PluginKeyValue, oldValue []byte
 	}
 
 	if oldValue == nil {
+		// Delete any existing, expired value.
+		if _, err := ps.GetMaster().Exec("DELETE FROM PluginKeyValueStore WHERE PluginId = :PluginId AND PKey = :Key AND ExpireAt != 0 AND ExpireAt < :CurrentTime",
+			map[string]interface{}{
+				"PluginId":    kv.PluginId,
+				"Key":         kv.Key,
+				"CurrentTime": model.GetMillis(),
+			}); err != nil {
+			return false, model.NewAppError("SqlPluginStore.CompareAndSet", "store.sql_plugin_store.delete.app_error", nil, err.Error(), http.StatusInternalServerError)
+		}
+
 		// Insert if oldValue is nil
 		if err := ps.GetMaster().Insert(kv); err != nil {
 			// If the error is from unique constraints violation, it's the result of a
@@ -100,15 +105,18 @@ func (ps SqlPluginStore) CompareAndSet(kv *model.PluginKeyValue, oldValue []byte
 			}
 		}
 	} else {
+		currentTime := model.GetMillis()
+
 		// Update if oldValue is not nil
 		updateResult, err := ps.GetMaster().Exec(
-			`UPDATE PluginKeyValueStore SET PValue = :New, ExpireAt = :ExpireAt WHERE PluginId = :PluginId AND PKey = :Key AND PValue = :Old`,
+			`UPDATE PluginKeyValueStore SET PValue = :New, ExpireAt = :ExpireAt WHERE PluginId = :PluginId AND PKey = :Key AND PValue = :Old AND (ExpireAt = 0 OR ExpireAt > :CurrentTime)`,
 			map[string]interface{}{
-				"PluginId": kv.PluginId,
+				"PluginId":    kv.PluginId,
-				"Key":      kv.Key,
+				"Key":         kv.Key,
-				"Old":      oldValue,
+				"Old":         oldValue,
-				"New":      kv.Value,
+				"New":         kv.Value,
-				"ExpireAt": kv.ExpireAt,
+				"ExpireAt":    kv.ExpireAt,
+				"CurrentTime": currentTime,
 			},
 		)
 		if err != nil {
@@ -126,11 +134,12 @@ func (ps SqlPluginStore) CompareAndSet(kv *model.PluginKeyValue, oldValue []byte
 				// atomicity. Nevertheless, let's return results consistent with Postgres and with what might
 				// be expected in this case.
 				count, err := ps.GetReplica().SelectInt(
-					"SELECT COUNT(*) FROM PluginKeyValueStore WHERE PluginId = :PluginId AND PKey = :Key AND PValue = :Value",
+					"SELECT COUNT(*) FROM PluginKeyValueStore WHERE PluginId = :PluginId AND PKey = :Key AND PValue = :Value AND (ExpireAt = 0 OR ExpireAt > :CurrentTime)",
 					map[string]interface{}{
-						"PluginId": kv.PluginId,
+						"PluginId":    kv.PluginId,
-						"Key":      kv.Key,
+						"Key":         kv.Key,
-						"Value":    kv.Value,
+						"Value":       kv.Value,
+						"CurrentTime": currentTime,
 					},
 				)
 				if err != nil {
@@ -166,11 +175,12 @@ func (ps SqlPluginStore) CompareAndDelete(kv *model.PluginKeyValue, oldValue []b
 	}
 
 	deleteResult, err := ps.GetMaster().Exec(
-		`DELETE FROM PluginKeyValueStore WHERE PluginId = :PluginId AND PKey = :Key AND PValue = :Old`,
+		`DELETE FROM PluginKeyValueStore WHERE PluginId = :PluginId AND PKey = :Key AND PValue = :Old AND (ExpireAt = 0 OR ExpireAt > :CurrentTime)`,
 		map[string]interface{}{
-			"PluginId": kv.PluginId,
+			"PluginId":    kv.PluginId,
-			"Key":      kv.Key,
+			"Key":         kv.Key,
-			"Old":      oldValue,
+			"Old":         oldValue,
+			"CurrentTime": model.GetMillis(),
 		},
 	)
 	if err != nil {
@@ -245,7 +255,7 @@ func (ps SqlPluginStore) DeleteAllExpired() *model.AppError {
 
 func (ps SqlPluginStore) List(pluginId string, offset int, limit int) ([]string, *model.AppError) {
 	if limit <= 0 {
-		limit = DEFAULT_PLUGIN_KEY_FETCH_LIMIT
+		limit = defaultPluginKeyFetchLimit
 	}
 
 	if offset <= 0 {
@@ -253,7 +263,7 @@ func (ps SqlPluginStore) List(pluginId string, offset int, limit int) ([]string,
 	}
 
 	var keys []string
-	_, err := ps.GetReplica().Select(&keys, "SELECT PKey FROM PluginKeyValueStore WHERE PluginId = :PluginId order by PKey limit :Limit offset :Offset", map[string]interface{}{"PluginId": pluginId, "Limit": limit, "Offset": offset})
+	_, err := ps.GetReplica().Select(&keys, "SELECT PKey FROM PluginKeyValueStore WHERE PluginId = :PluginId AND (ExpireAt = 0 OR ExpireAt > :CurrentTime) order by PKey limit :Limit offset :Offset", map[string]interface{}{"PluginId": pluginId, "Limit": limit, "Offset": offset, "CurrentTime": model.GetMillis()})
 	if err != nil {
 		return nil, model.NewAppError("SqlPluginStore.List", "store.sql_plugin_store.list.app_error", nil, fmt.Sprintf("plugin_id=%v, err=%v", pluginId, err.Error()), http.StatusInternalServerError)
 	}

store/sqlstore/plugin_store_test.go

@@ -10,5 +10,5 @@ import (
 )
 
 func TestPluginStore(t *testing.T) {
-	StoreTest(t, storetest.TestPluginStore)
+	StoreTestWithSqlSupplier(t, storetest.TestPluginStore)
 }