core/storage: fix nil data unmarshal

pkg/storage/postgres/backend_test.go

@@ -51,18 +51,20 @@ func TestBackend(t *testing.T) {
 		})
 
 		t.Run("delete", func(t *testing.T) {
-			serverVersion, err := backend.Put(ctx, []*databroker.Record{
-				{
-					Type: "test-1",
-					Id:   "r3",
-					Data: protoutil.NewAny(protoutil.NewStructMap(map[string]*structpb.Value{
-						"k1": protoutil.NewStructString("v1"),
-					})),
-					DeletedAt: timestamppb.Now(),
-				},
-			})
+			serverVersion, err := backend.Put(ctx, []*databroker.Record{{
+				Type:      "test-1",
+				Id:        "r3",
+				DeletedAt: timestamppb.Now(),
+			}})
 			assert.NotEqual(t, 0, serverVersion)
 			assert.NoError(t, err)
+
+			stream, err := backend.Sync(ctx, "test-1", serverVersion, 0)
+			require.NoError(t, err)
+			t.Cleanup(func() { _ = stream.Close() })
+			records, err := storage.RecordStreamToList(stream)
+			require.NoError(t, err)
+			assert.NotEmpty(t, records)
 		})
 
 		t.Run("capacity", func(t *testing.T) {

pkg/storage/postgres/postgres.go

@@ -126,13 +126,17 @@ func getNextChangedRecord(ctx context.Context, q querier, recordType string, aft
 		return nil, fmt.Errorf("error querying next changed record: %w", err)
 	}
 
-	a, err := protoutil.UnmarshalAnyJSON(data)
-	if isUnknownType(err) {
-		a = protoutil.ToAny(protoutil.ToStruct(map[string]string{
-			"id": recordID,
-		}))
-	} else if err != nil {
-		return nil, fmt.Errorf("error unmarshaling changed record data: %w", err)
+	// data may be nil if a record is deleted
+	var a *anypb.Any
+	if len(data) != 0 {
+		a, err = protoutil.UnmarshalAnyJSON(data)
+		if isUnknownType(err) {
+			a = protoutil.ToAny(protoutil.ToStruct(map[string]string{
+				"id": recordID,
+			}))
+		} else if err != nil {
+			return nil, fmt.Errorf("error unmarshaling changed record data: %w", err)
+		}
 	}
 
 	return &databroker.Record{