Fix dynamodbbk's interactions with item expiry (#68038)

* Fix CRUD expiry behavior in dynamodbbk

* Fix AtomicWrite expiry behavior in dynamodbbk

* DynamoDB relies on server-side TTL expiry so it can't use a fake clock

* fixup! Fix CRUD expiry behavior in dynamodbbk

* Allow Delete to delete expired items

* Remove some unnecessary whitespace

* Take the time before the backend call in Get and Delete

lib/backend/dynamo/atomicwrite.go

@@ -42,19 +42,13 @@ const (
 	txnAttemptLogInterval = 8
 )
 
-var (
-	existsExpr          = "attribute_exists(FullPath)"
-	notExistsExpr       = "attribute_not_exists(FullPath)"
-	revisionExpr        = "Revision = :rev AND attribute_exists(FullPath)"
-	missingRevisionExpr = "attribute_not_exists(Revision) AND attribute_exists(FullPath)"
-)
-
 func (b *Backend) AtomicWrite(ctx context.Context, condacts []backend.ConditionalAction) (revision string, err error) {
 	if err := backend.ValidateAtomicWrite(condacts); err != nil {
 		return "", trace.Wrap(err)
 	}
 
 	revision = backend.CreateRevision()
+	now := b.clock.Now()
 
 	tableName := aws.String(b.TableName)
 
@@ -69,9 +63,15 @@ func (b *Backend) AtomicWrite(ctx context.Context, condacts []backend.Conditiona
 		case backend.KindWhatever:
 			// no comparison to assert
 		case backend.KindExists:
-			condExpr = &existsExpr
+			condExpr = aws.String("attribute_exists(FullPath) AND (attribute_not_exists(Expires) OR Expires >= :now)")
+			exprAttrValues = map[string]types.AttributeValue{
+				":now": timeToAttributeValue(now),
+			}
 		case backend.KindNotExists:
-			condExpr = &notExistsExpr
+			condExpr = aws.String("attribute_not_exists(FullPath) OR Expires < :now")
+			exprAttrValues = map[string]types.AttributeValue{
+				":now": timeToAttributeValue(now),
+			}
 		case backend.KindRevision:
 			switch ca.Condition.Revision {
 			case "":
@@ -79,44 +79,39 @@ func (b *Backend) AtomicWrite(ctx context.Context, condacts []backend.Conditiona
 				return "", trace.Wrap(backend.ErrConditionFailed)
 			case backend.BlankRevision:
 				// item has not been modified since the introduction of the revision attr
-				condExpr = &missingRevisionExpr
+				condExpr = aws.String("attribute_exists(FullPath) AND attribute_not_exists(Revision) AND (attribute_not_exists(Expires) OR Expires >= :now)")
+				exprAttrValues = map[string]types.AttributeValue{
+					":now": timeToAttributeValue(now),
+				}
 			default:
 				// revision is expected to be present and well-defined
-				condExpr = &revisionExpr
+				condExpr = aws.String("Revision = :rev AND (attribute_not_exists(Expires) OR Expires >= :now)")
 				exprAttrValues = map[string]types.AttributeValue{
 					":rev": &types.AttributeValueMemberS{Value: ca.Condition.Revision},
+					":now": timeToAttributeValue(now),
 				}
 			}
 		default:
-			return "", trace.BadParameter("unexpected condition kind %v in conditional action against key %q", ca.Condition.Kind, ca.Key)
+			return "", trace.BadParameter("unexpected condition kind %v in conditional action against key %+q", ca.Condition.Kind, ca.Key.String())
 		}
 
-		fullPath := prependPrefix(ca.Key)
-
 		var txnItem types.TransactWriteItem
 
 		switch ca.Action.Kind {
 		case backend.KindNop:
-			av, err := attributevalue.MarshalMap(keyLookup{
-				HashKey:  hashKey,
-				FullPath: fullPath,
-			})
-			if err != nil {
-				return "", trace.Wrap(err)
-			}
-
 			txnItem.ConditionCheck = &types.ConditionCheck{
+				TableName: tableName,
+				Key:       keyToAttributeValueMap(ca.Key),
+
 				ConditionExpression:       condExpr,
 				ExpressionAttributeValues: exprAttrValues,
-				Key:                       av,
-				TableName:                 tableName,
 			}
 
 		case backend.KindPut:
 			includesPut = true
 			r := record{
 				HashKey:   hashKey,
-				FullPath:  fullPath,
+				FullPath:  prependPrefix(ca.Key),
 				Value:     ca.Action.Item.Value,
 				Timestamp: time.Now().UTC().Unix(),
 				Revision:  revision,
@@ -131,29 +126,24 @@ func (b *Backend) AtomicWrite(ctx context.Context, condacts []backend.Conditiona
 			}
 
 			txnItem.Put = &types.Put{
+				TableName: tableName,
+
+				Item: av,
+
 				ConditionExpression:       condExpr,
 				ExpressionAttributeValues: exprAttrValues,
-				Item:                      av,
-				TableName:                 tableName,
 			}
 		case backend.KindDelete:
-			av, err := attributevalue.MarshalMap(keyLookup{
-				HashKey:  hashKey,
-				FullPath: fullPath,
-			})
-			if err != nil {
-				return "", trace.Wrap(err)
-			}
-
 			txnItem.Delete = &types.Delete{
+				TableName: tableName,
+				Key:       keyToAttributeValueMap(ca.Key),
+
 				ConditionExpression:       condExpr,
 				ExpressionAttributeValues: exprAttrValues,
-				Key:                       av,
-				TableName:                 tableName,
 			}
 
 		default:
-			return "", trace.BadParameter("unexpected action kind %v in conditional action against key %q", ca.Action.Kind, ca.Key)
+			return "", trace.BadParameter("unexpected action kind %v in conditional action against key %+q", ca.Action.Kind, ca.Key.String())
 		}
 
 		txnItems = append(txnItems, txnItem)

lib/backend/dynamo/dynamodbbk.go

@@ -25,6 +25,7 @@ import (
 	"log/slog"
 	"net/http"
 	"strconv"
+	"strings"
 	"sync/atomic"
 	"time"
 
@@ -186,11 +187,6 @@ type record struct {
 	Timestamp int64
 }
 
-type keyLookup struct {
-	HashKey  string
-	FullPath string
-}
-
 const (
 	// hashKey is actually the name of the partition. This backend
 	// places all objects in the same DynamoDB partition
@@ -517,7 +513,7 @@ func (b *Backend) GetName() string {
 func (b *Backend) Create(ctx context.Context, item backend.Item) (*backend.Lease, error) {
 	rev, err := b.create(ctx, item, modeCreate)
 	if trace.IsCompareFailed(err) {
-		err = trace.AlreadyExists("%s", err)
+		err = trace.AlreadyExists("%+q already exists", item.Key.String())
 	}
 	if err != nil {
 		return nil, trace.Wrap(err)
@@ -541,7 +537,7 @@ func (b *Backend) Put(ctx context.Context, item backend.Item) (*backend.Lease, e
 func (b *Backend) Update(ctx context.Context, item backend.Item) (*backend.Lease, error) {
 	rev, err := b.create(ctx, item, modeUpdate)
 	if trace.IsCompareFailed(err) {
-		err = trace.NotFound("%s", err)
+		err = trace.NotFound("%+q is not found", item.Key.String())
 	}
 	if err != nil {
 		return nil, trace.Wrap(err)
@@ -602,13 +598,13 @@ func (b *Backend) Items(ctx context.Context, params backend.ItemsParams) iter.Se
 
 		// filter out expired items, otherwise they might show up in the query
 		// http://docs.aws.amazon.com/amazondynamodb/latest/developerguide/howitworks-ttl.html
-		filter = "attribute_not_exists(Expires) OR Expires >= :timestamp"
+		filter = "attribute_not_exists(Expires) OR Expires >= :now"
 	)
 
 	av := map[string]types.AttributeValue{
 		":rangeStart": &types.AttributeValueMemberS{Value: prependPrefix(params.StartKey)},
 		":rangeEnd":   &types.AttributeValueMemberS{Value: prependPrefix(params.EndKey)},
-		":timestamp":  timeToAttributeValue(b.clock.Now().UTC()),
+		":now":        timeToAttributeValue(b.clock.Now()),
 		":hashKey":    &types.AttributeValueMemberS{Value: hashKey},
 	}
 
@@ -760,10 +756,7 @@ func (b *Backend) DeleteRange(ctx context.Context, startKey, endKey backend.Key)
 
 		requests = append(requests, types.WriteRequest{
 			DeleteRequest: &types.DeleteRequest{
-				Key: map[string]types.AttributeValue{
-					hashKeyKey:  &types.AttributeValueMemberS{Value: hashKey},
-					fullPathKey: &types.AttributeValueMemberS{Value: prependPrefix(item.Key)},
-				},
+				Key: keyToAttributeValueMap(item.Key),
 			},
 		})
 
@@ -806,7 +799,7 @@ func (b *Backend) Get(ctx context.Context, key backend.Key) (*backend.Item, erro
 		Revision: r.Revision,
 	}
 	if r.Expires != nil {
-		item.Expires = time.Unix(*r.Expires, 0)
+		item.Expires = time.Unix(*r.Expires, 0).UTC()
 	}
 
 	if item.Revision == "" {
@@ -844,24 +837,26 @@ func (b *Backend) CompareAndSwap(ctx context.Context, expected backend.Item, rep
 	if err != nil {
 		return nil, trace.Wrap(err)
 	}
-	input := dynamodb.PutItemInput{
-		Item:                av,
-		TableName:           aws.String(b.TableName),
-		ConditionExpression: aws.String("#v = :prev"),
+	input := &dynamodb.PutItemInput{
+		TableName: aws.String(b.TableName),
+
+		Item: av,
+
+		ConditionExpression: aws.String("#v = :prev AND (attribute_not_exists(Expires) OR Expires >= :now)"),
 		ExpressionAttributeNames: map[string]string{
 			"#v": "Value",
 		},
 		ExpressionAttributeValues: map[string]types.AttributeValue{
 			":prev": &types.AttributeValueMemberB{Value: expected.Value},
+			":now":  timeToAttributeValue(b.clock.Now()),
 		},
 	}
 
-	_, err = b.svc.PutItem(ctx, &input)
+	_, err = b.svc.PutItem(ctx, input)
 	err = convertError(err)
 	if err != nil {
-		// in this case let's use more specific compare failed error
-		if trace.IsAlreadyExists(err) {
-			return nil, trace.CompareFailed("%s", err)
+		if trace.IsCompareFailed(err) {
+			return nil, trace.CompareFailed("%+q not found or does not match expected", replaceWith.Key.String())
 		}
 		return nil, trace.Wrap(err)
 	}
@@ -870,10 +865,36 @@ func (b *Backend) CompareAndSwap(ctx context.Context, expected backend.Item, rep
 
 // Delete deletes item by key
 func (b *Backend) Delete(ctx context.Context, key backend.Key) error {
-	if _, err := b.getKey(ctx, key); err != nil {
-		return err
+	// since dynamodb's built-in deletion of expired items can be very slow,
+	// instead of filtering the point delete here to only delete if the item is
+	// not expired we unconditionally delete the item, and we return the correct
+	// error by checking if the item that was just deleted was expired
+	now := b.clock.Now().Truncate(time.Second)
+	out, err := b.svc.DeleteItem(ctx, &dynamodb.DeleteItemInput{
+		TableName: aws.String(b.TableName),
+		Key:       keyToAttributeValueMap(key),
+
+		ReturnValues: types.ReturnValueAllOld,
+	})
+	if err != nil {
+		return trace.Wrap(err)
 	}
-	return b.deleteKey(ctx, key)
+
+	if len(out.Attributes) < 1 {
+		return trace.NotFound("%+q is not found", key.String())
+	}
+
+	var r struct {
+		Expires *attributevalue.UnixTime
+	}
+	if err := attributevalue.UnmarshalMap(out.Attributes, &r); err != nil {
+		return trace.Wrap(err, "checking expiry")
+	}
+	if r.Expires != nil && time.Time(*r.Expires).Before(now) {
+		return trace.NotFound("%+q is not found", key.String())
+	}
+
+	return nil
 }
 
 // ConditionalUpdate updates the matching item in Dynamo if the provided revision matches
@@ -889,6 +910,9 @@ func (b *Backend) ConditionalUpdate(ctx context.Context, item backend.Item) (*ba
 
 	rev, err := b.create(ctx, item, modeConditionalUpdate)
 	if err != nil {
+		if trace.IsCompareFailed(err) {
+			err = trace.Wrap(backend.ErrIncorrectRevision)
+		}
 		return nil, trace.Wrap(err)
 	}
 
@@ -903,27 +927,25 @@ func (b *Backend) ConditionalDelete(ctx context.Context, key backend.Key, rev st
 		return trace.Wrap(backend.ErrIncorrectRevision)
 	}
 
-	av, err := attributevalue.MarshalMap(keyLookup{
-		HashKey:  hashKey,
-		FullPath: prependPrefix(key),
-	})
-	if err != nil {
-		return trace.Wrap(err)
-	}
-
-	input := dynamodb.DeleteItemInput{
-		Key:       av,
+	input := &dynamodb.DeleteItemInput{
 		TableName: aws.String(b.TableName),
+		Key:       keyToAttributeValueMap(key),
 	}
 
 	if rev == backend.BlankRevision {
-		input.ConditionExpression = aws.String("attribute_not_exists(Revision) AND attribute_exists(FullPath)")
+		input.ConditionExpression = aws.String("attribute_exists(FullPath) AND attribute_not_exists(Revision) AND (attribute_not_exists(Expires) OR Expires >= :now)")
+		input.ExpressionAttributeValues = map[string]types.AttributeValue{
+			":now": timeToAttributeValue(b.clock.Now()),
+		}
 	} else {
-		input.ExpressionAttributeValues = map[string]types.AttributeValue{":rev": &types.AttributeValueMemberS{Value: rev}}
-		input.ConditionExpression = aws.String("Revision = :rev AND attribute_exists(FullPath)")
+		input.ConditionExpression = aws.String("Revision = :rev AND (attribute_not_exists(Expires) OR Expires >= :now)")
+		input.ExpressionAttributeValues = map[string]types.AttributeValue{
+			":rev": &types.AttributeValueMemberS{Value: rev},
+			":now": timeToAttributeValue(b.clock.Now()),
+		}
 	}
 
-	if _, err = b.svc.DeleteItem(ctx, &input); err != nil {
+	if _, err := b.svc.DeleteItem(ctx, input); err != nil {
 		err = convertError(err)
 		if trace.IsCompareFailed(err) {
 			return trace.Wrap(backend.ErrIncorrectRevision)
@@ -947,22 +969,20 @@ func (b *Backend) KeepAlive(ctx context.Context, lease backend.Lease, expires ti
 		return trace.BadParameter("lease is missing key")
 	}
 	input := &dynamodb.UpdateItemInput{
-		ExpressionAttributeValues: map[string]types.AttributeValue{
-			":expires":   &types.AttributeValueMemberN{Value: strconv.FormatInt(expires.UTC().Unix(), 10)},
-			":timestamp": &types.AttributeValueMemberN{Value: strconv.FormatInt(b.clock.Now().UTC().Unix(), 10)},
-		},
 		TableName: aws.String(b.TableName),
-		Key: map[string]types.AttributeValue{
-			hashKeyKey:  &types.AttributeValueMemberS{Value: hashKey},
-			fullPathKey: &types.AttributeValueMemberS{Value: prependPrefix(lease.Key)},
-		},
+		Key:       keyToAttributeValueMap(lease.Key),
+
 		UpdateExpression:    aws.String("SET Expires = :expires"),
-		ConditionExpression: aws.String("attribute_exists(FullPath) AND (attribute_not_exists(Expires) OR Expires >= :timestamp)"),
+		ConditionExpression: aws.String("attribute_exists(FullPath) AND (attribute_not_exists(Expires) OR Expires >= :now)"),
+		ExpressionAttributeValues: map[string]types.AttributeValue{
+			":expires": timeToAttributeValue(expires),
+			":now":     timeToAttributeValue(b.clock.Now()),
+		},
 	}
 	_, err := b.svc.UpdateItem(ctx, input)
 	err = convertError(err)
 	if trace.IsCompareFailed(err) {
-		err = trace.NotFound("%s", err)
+		err = trace.NotFound("%+q is not found", lease.Key.String())
 	}
 	return err
 }
@@ -1096,16 +1116,6 @@ func (b *Backend) createTable(ctx context.Context, tableName *string, rangeKey s
 	return trace.Wrap(err)
 }
 
-// isExpired returns 'true' if the given object (record) has a TTL and
-// it's due.
-func (r *record) isExpired(now time.Time) bool {
-	if r.Expires == nil {
-		return false
-	}
-	expiryDateUTC := time.Unix(*r.Expires, 0).UTC()
-	return now.UTC().After(expiryDateUTC)
-}
-
 const (
 	modeCreate = iota
 	modePut
@@ -1121,7 +1131,7 @@ func prependPrefix(key backend.Key) string {
 
 // trimPrefix removes leading 'teleport' from the key
 func trimPrefix(key string) backend.Key {
-	return backend.KeyFromString(key).TrimPrefix(backend.KeyFromString(keyPrefix))
+	return backend.KeyFromString(strings.TrimPrefix(key, keyPrefix))
 }
 
 // create is a helper that writes a key/value pair in Dynamo with a given expiration.
@@ -1144,105 +1154,77 @@ func (b *Backend) create(ctx context.Context, item backend.Item, mode int) (stri
 		return "", trace.Wrap(err)
 	}
 	input := dynamodb.PutItemInput{
-		Item:      av,
 		TableName: aws.String(b.TableName),
+
+		Item: av,
 	}
 
 	switch mode {
 	case modeCreate:
-		input.ConditionExpression = aws.String("attribute_not_exists(FullPath)")
+		input.ConditionExpression = aws.String("attribute_not_exists(FullPath) OR Expires < :now")
+		input.ExpressionAttributeValues = map[string]types.AttributeValue{
+			":now": timeToAttributeValue(b.clock.Now()),
+		}
 	case modeUpdate:
-		input.ConditionExpression = aws.String("attribute_exists(FullPath)")
+		input.ConditionExpression = aws.String("attribute_exists(FullPath) AND (attribute_not_exists(Expires) OR Expires >= :now)")
+		input.ExpressionAttributeValues = map[string]types.AttributeValue{
+			":now": timeToAttributeValue(b.clock.Now()),
+		}
 	case modePut:
 	case modeConditionalUpdate:
 		// If the revision is empty, then the resource existed prior to revision support. Instead of validating that
 		// the revisions match, validate that the revision attribute does not exist. Otherwise, validate that the revision
 		// attribute matches the item revision.
 		if item.Revision == "" {
-			input.ConditionExpression = aws.String("attribute_not_exists(Revision) AND attribute_exists(FullPath)")
+			input.ConditionExpression = aws.String("attribute_exists(FullPath) AND attribute_not_exists(Revision) AND (attribute_not_exists(Expires) OR Expires >= :now)")
+			input.ExpressionAttributeValues = map[string]types.AttributeValue{
+				":now": timeToAttributeValue(b.clock.Now()),
+			}
 		} else {
-			input.ExpressionAttributeValues = map[string]types.AttributeValue{":rev": &types.AttributeValueMemberS{Value: item.Revision}}
-			input.ConditionExpression = aws.String("Revision = :rev AND attribute_exists(FullPath)")
+			input.ConditionExpression = aws.String("Revision = :rev AND (attribute_not_exists(Expires) OR Expires >= :now)")
+			input.ExpressionAttributeValues = map[string]types.AttributeValue{
+				":rev": &types.AttributeValueMemberS{Value: item.Revision},
+				":now": timeToAttributeValue(b.clock.Now()),
+			}
 		}
 	default:
-		return "", trace.BadParameter("unrecognized mode")
+		return "", trace.BadParameter("unrecognized write mode %d (this is a bug)", mode)
 	}
 	_, err = b.svc.PutItem(ctx, &input)
 	err = convertError(err)
 	if err != nil {
-		if mode == modeConditionalUpdate && trace.IsCompareFailed(err) {
-			return "", trace.Wrap(backend.ErrIncorrectRevision)
-		}
-
 		return "", trace.Wrap(err)
 	}
 
 	return r.Revision, nil
 }
 
-func (b *Backend) deleteKey(ctx context.Context, key backend.Key) error {
-	av, err := attributevalue.MarshalMap(keyLookup{
-		HashKey:  hashKey,
-		FullPath: prependPrefix(key),
-	})
-	if err != nil {
-		return trace.Wrap(err)
-	}
-	input := dynamodb.DeleteItemInput{Key: av, TableName: aws.String(b.TableName)}
-	if _, err = b.svc.DeleteItem(ctx, &input); err != nil {
-		return trace.Wrap(err)
-	}
-	return nil
-}
-
-func (b *Backend) deleteKeyIfExpired(ctx context.Context, key backend.Key) error {
-	_, err := b.svc.DeleteItem(ctx, &dynamodb.DeleteItemInput{
+func (b *Backend) getKey(ctx context.Context, key backend.Key) (*record, error) {
+	input := &dynamodb.GetItemInput{
 		TableName: aws.String(b.TableName),
 		Key:       keyToAttributeValueMap(key),
 
-		// succeed if the item no longer exists
-		ConditionExpression: aws.String(
-			"attribute_not_exists(FullPath) OR (attribute_exists(Expires) AND Expires <= :timestamp)",
-		),
-		ExpressionAttributeValues: map[string]types.AttributeValue{
-			":timestamp": timeToAttributeValue(b.clock.Now()),
-		},
-	})
-	return trace.Wrap(err)
-}
-
-func (b *Backend) getKey(ctx context.Context, key backend.Key) (*record, error) {
-	av, err := attributevalue.MarshalMap(keyLookup{
-		HashKey:  hashKey,
-		FullPath: prependPrefix(key),
-	})
-	if err != nil {
-		return nil, trace.Wrap(err)
-	}
-	input := dynamodb.GetItemInput{
-		Key:            av,
-		TableName:      aws.String(b.TableName),
 		ConsistentRead: aws.Bool(true),
 	}
-	out, err := b.svc.GetItem(ctx, &input)
+	now := b.clock.Now().Truncate(time.Second)
+	out, err := b.svc.GetItem(ctx, input)
 	if err != nil {
 		// we deliberately use a "generic" trace error here, since we don't want
 		// callers to make assumptions about the nature of the failure.
-		return nil, trace.WrapWithMessage(err, "failed to get %q (dynamo error)", key.String())
+		return nil, trace.WrapWithMessage(err, "failed to get %+q (dynamo error)", key.String())
 	}
 	if len(out.Item) == 0 {
-		return nil, trace.NotFound("%q is not found", key.String())
+		return nil, trace.NotFound("%+q is not found", key.String())
 	}
 	var r record
 	if err := attributevalue.UnmarshalMap(out.Item, &r); err != nil {
-		return nil, trace.WrapWithMessage(err, "failed to unmarshal dynamo item %q", key.String())
+		return nil, trace.WrapWithMessage(err, "failed to unmarshal dynamo item %+q", key.String())
 	}
-	// Check if key expired, if expired delete it
-	if r.isExpired(b.clock.Now()) {
-		if err := b.deleteKeyIfExpired(ctx, key); err != nil {
-			b.logger.WarnContext(ctx, "Failed deleting expired key", "key", key, "error", err)
-		}
-		return nil, trace.NotFound("%q is not found", key.String())
+	// item expiry is stored as integer seconds since the unix epoch and so are
+	// the time filters in queries, so we replicate the same behavior here by
+	// comparing the expiry against the current clock truncated to the second
+	if r.Expires != nil && time.Unix(*r.Expires, 0).Before(now) {
+		return nil, trace.NotFound("%+q is not found", key.String())
 	}
 	return &r, nil
 }
@@ -1299,17 +1281,13 @@ func convertError(err error) error {
 	return err
 }
 
-func fullPathToAttributeValueMap(fullPath string) map[string]types.AttributeValue {
+func keyToAttributeValueMap(key backend.Key) map[string]types.AttributeValue {
 	return map[string]types.AttributeValue{
 		hashKeyKey:  &types.AttributeValueMemberS{Value: hashKey},
-		fullPathKey: &types.AttributeValueMemberS{Value: fullPath},
+		fullPathKey: &types.AttributeValueMemberS{Value: prependPrefix(key)},
 	}
 }
 
-func keyToAttributeValueMap(key backend.Key) map[string]types.AttributeValue {
-	return fullPathToAttributeValueMap(prependPrefix(key))
-}
-
 func timeToAttributeValue(t time.Time) types.AttributeValue {
 	return &types.AttributeValueMemberN{
 		Value: strconv.FormatInt(t.Unix(), 10),

lib/backend/dynamo/dynamodbbk_test.go

@@ -415,9 +415,7 @@ func TestDynamoDB(t *testing.T) {
 		if err != nil {
 			return nil, nil, trace.Wrap(err)
 		}
-		clock := clockwork.NewFakeClockAt(time.Now())
-		uut.clock = clock
-		return uut, clock, nil
+		return uut, test.BlockingFakeClock{Clock: clockwork.NewRealClock()}, nil
 	}
 
 	test.RunBackendComplianceSuite(t, newBackend)

lib/backend/dynamo/shards.go

@@ -394,7 +394,7 @@ func toEvent(rec streamtypes.Record) (*backend.Event, error) {
 		}
 		var expires time.Time
 		if r.Expires != nil {
-			expires = time.Unix(*r.Expires, 0)
+			expires = time.Unix(*r.Expires, 0).UTC()
 		}
 		return &backend.Event{
 			Type: op,