Fixes duplicate idempotent key issue during recovery and non-recovery

internal/sessions/committable/committable.go

@@ -141,15 +141,14 @@ func (a *SavepointCommittable) Commit(ctx context.Context, idempotencyKey []byte
 	a.writable = false
 
 	if a.skip {
+		a.skip = false
 		return a.db.Get(ctx, ApphashKey, true)
 	}
 
 	if a.savepoint == nil {
 		return nil, fmt.Errorf("no session exists")
 	}
 
-	a.skip = false
-
 	var appHash []byte
 	if a.idFn != nil {
 		var err error

internal/sessions/kv.go

@@ -15,11 +15,6 @@ func setCurrentKey(kv KV, idempotencyKey []byte) error {
 	return kv.Set(currentKeyKey, idempotencyKey)
 }
 
-// deleteCurrentKey deletes the current key in the given connection.
-func deleteCurrentKey(kv KV) error {
-	return kv.Delete(currentKeyKey)
-}
-
 // getCurrentKey gets the current key in the given connection.
 // it can return nil if there is no current key.
 func getCurrentKey(kv KV) ([]byte, error) {

internal/sessions/session.go

@@ -79,11 +79,9 @@ func (m *MultiCommitter) Begin(ctx context.Context, idempotencyKey []byte) (err
 		return err
 	}
 
-	// if lastKey is not nil, we are recovering from a crash
-	if lastKey != nil {
-		if !bytes.Equal(lastKey, idempotencyKey) {
-			return fmt.Errorf("%w on recovery: expected %s, got %s", ErrIdempotencyKeyMismatch, lastKey, idempotencyKey)
-		}
+	// if the last key is the same as the current key, we are recovering
+	// from a crash, therefore begin recovery mode.
+	if bytes.Equal(lastKey, idempotencyKey) {
 		for _, committable := range m.committables {
 			err = committable.BeginRecovery(ctx, idempotencyKey)
 			if err != nil {
@@ -143,7 +141,6 @@ func (m *MultiCommitter) Commit(ctx context.Context, idempotencyKey []byte) (id
 		id = append(id, newId...)
 	}
 
-	err = deleteCurrentKey(m.kv)
 	return id, err
 }
 

internal/sessions/session_test.go

@@ -45,7 +45,8 @@ func Test_Sessions(t *testing.T) {
 				_, err = mc.Commit(ctx, key2)
 				assert.NoError(t, err)
 
-				assert.Equal(t, len(kv.vals), 0)
+				// IdempotentKey is never deleted, just updated with every block, so it should always be 1
+				assert.Equal(t, len(kv.vals), 1)
 			},
 		},
 		{