fix(cart): make cart mutations atomic and stop checkout from wiping concurrent adds

Update ValkeyCartStore to perform cart mutations via Redis optimistic concurrency transactions, avoiding lost updates during concurrent AddItem requests. Update checkout service to only remove ordered items from the cart instead of unconditionally wiping it. Update cart-unit-tests CI workflow to start valkey via docker compose.

Signed-off-by: bhuvan-somisetty <somisettybhuvan5@gmail.com>

Author: bhuvan-somisetty

.github/workflows/checks.yml

@@ -145,6 +145,34 @@ jobs:
             "${WEAVER_IMAGE}" \
             registry check -r source
 
+  cart-unit-tests:
+    name: Cart unit tests
+    runs-on: ubuntu-latest
+    steps:
+      - name: Checkout code
+        uses: actions/checkout@3d3c42e5aac5ba805825da76410c181273ba90b1  # v7.0.1
+        with:
+          persist-credentials: false
+      - name: Start Valkey
+        run: |
+          VALKEY_IMAGE=$(grep -oP '(?<=^VALKEY_IMAGE=)\S*' .env)
+          docker run -d --rm --name valkey -p 6379:6379 "${VALKEY_IMAGE}"
+          for i in {1..30}; do
+            if docker exec valkey valkey-cli ping 2>/dev/null | grep -q PONG; then
+              break
+            fi
+            sleep 1
+          done
+      - name: Set up .NET
+        uses: actions/setup-dotnet@a98b56852c35b8e3190ac28c8c2271da59106c68  # v6.0.0
+        with:
+          dotnet-version: '10.0.x'
+      - name: Run cart unit tests
+        working-directory: src/cart
+        env:
+          VALKEY_ADDR: localhost:6379
+        run: dotnet test tests/cart.tests.csproj
+
   check-react-native-changes:
     name: Check React Native changes
     runs-on: ubuntu-latest
@@ -190,6 +218,7 @@ jobs:
       sanity,
       checklicense,
       weaver-check,
+      cart-unit-tests,
       react-native-build,
       codeql-analysis
     ]

CHANGELOG.md

@@ -19,6 +19,9 @@ the release.
   v1.46.0/v0.71.0 release line, picking up otelgrpc recording `error.type`
   on RPC duration metrics for failed calls
   ([#3901](https://github.com/open-telemetry/opentelemetry-demo/issues/3901))
+* [cart] Make cart mutations atomic using Redis optimistic concurrency
+  transactions, and update checkout to only remove ordered items
+  ([#3825](https://github.com/open-telemetry/opentelemetry-demo/pull/3825))
 * [collector] Add a data redaction/deletion example: delete, hash, and partially
   mask sensitive attributes with the transform processor, plus a documented
   redaction-processor fallback

src/cart/src/cartstore/ValkeyCartStore.cs

@@ -18,6 +18,7 @@ public class ValkeyCartStore : ICartStore
     private readonly ILogger _logger;
     private const string CartFieldName = "cart";
     private const int RedisRetryNumber = 30;
+    private const int CartMutationMaxAttempts = 30;
 
     private volatile ConnectionMultiplexer _redis;
     private volatile bool _isRedisConnectionOpened;
@@ -132,38 +133,28 @@ public async Task AddItemAsync(string userId, string productId, int quantity)
 
         try
         {
-            EnsureRedisConnected();
-
-            var db = _redis.GetDatabase();
-
-            // Access the cart from the cache
-            var value = await db.HashGetAsync(userId, CartFieldName);
-
-            Oteldemo.Cart cart;
-            if (value.IsNull)
+            await MutateCartAsync(userId, cart =>
             {
-                cart = new Oteldemo.Cart
-                {
-                    UserId = userId
-                };
-                cart.Items.Add(new Oteldemo.CartItem { ProductId = productId, Quantity = quantity });
-            }
-            else
-            {
-                cart = Oteldemo.Cart.Parser.ParseFrom((byte[])value);
                 var existingItem = cart.Items.SingleOrDefault(i => i.ProductId == productId);
                 if (existingItem == null)
                 {
-                    cart.Items.Add(new Oteldemo.CartItem { ProductId = productId, Quantity = quantity });
+                    if (quantity > 0)
+                    {
+                        cart.Items.Add(new Oteldemo.CartItem { ProductId = productId, Quantity = quantity });
+                    }
+                    return;
                 }
-                else
+
+                existingItem.Quantity += quantity;
+                if (existingItem.Quantity <= 0)
                 {
-                    existingItem.Quantity += quantity;
+                    cart.Items.Remove(existingItem);
                 }
-            }
-
-            await db.HashSetAsync(userId, new[]{ new HashEntry(CartFieldName, cart.ToByteArray()) });
-            await db.KeyExpireAsync(userId, TimeSpan.FromMinutes(60));
+            });
+        }
+        catch (RpcException)
+        {
+            throw;
         }
         catch (Exception ex)
         {
@@ -175,6 +166,42 @@ public async Task AddItemAsync(string userId, string productId, int quantity)
         }
     }
 
+    private async Task MutateCartAsync(string userId, Action<Oteldemo.Cart> mutate)
+    {
+        EnsureRedisConnected();
+
+        var db = _redis.GetDatabase();
+
+        for (var attempt = 0; attempt < CartMutationMaxAttempts; attempt++)
+        {
+            var existingValue = await db.HashGetAsync(userId, CartFieldName);
+
+            var cart = existingValue.IsNull
+                ? new Oteldemo.Cart { UserId = userId }
+                : Oteldemo.Cart.Parser.ParseFrom((byte[])existingValue);
+
+            mutate(cart);
+
+            var transaction = db.CreateTransaction();
+            transaction.AddCondition(existingValue.IsNull
+                ? Condition.HashNotExists(userId, CartFieldName)
+                : Condition.HashEqual(userId, CartFieldName, existingValue));
+
+            _ = transaction.HashSetAsync(userId, new[] { new HashEntry(CartFieldName, cart.ToByteArray()) });
+            _ = transaction.KeyExpireAsync(userId, TimeSpan.FromMinutes(60));
+
+            if (await transaction.ExecuteAsync())
+            {
+                return;
+            }
+
+            await Task.Delay(Random.Shared.Next(2, 10) * (attempt + 1));
+        }
+
+        throw new RpcException(new Status(StatusCode.Aborted,
+            $"Couldn't update cart for user {userId} after {CartMutationMaxAttempts} attempts due to concurrent modifications."));
+    }
+
     public async Task EmptyCartAsync(string userId)
     {
         Log.EmptyCartAsync(_logger, userId);

src/cart/tests/CartServiceTests.cs

@@ -1,12 +1,20 @@
 // Copyright The OpenTelemetry Authors
 // SPDX-License-Identifier: Apache-2.0
 using System;
+using System.Collections.Generic;
+using System.Linq;
 using System.Threading.Tasks;
 using Grpc.Net.Client;
 using Oteldemo;
+using Microsoft.AspNetCore.Builder;
+using Microsoft.AspNetCore.Hosting;
 using Microsoft.AspNetCore.TestHost;
+using Microsoft.Extensions.DependencyInjection;
 using Microsoft.Extensions.Hosting;
+using Microsoft.Extensions.Logging.Abstractions;
+using OpenFeature;
 using Xunit;
+using cart.cartstore;
 using static Oteldemo.CartService;
 
 namespace cart.tests;
@@ -17,59 +25,68 @@ public class CartServiceTests
 
     public CartServiceTests()
     {
+        var valkeyAddress = Environment.GetEnvironmentVariable("VALKEY_ADDR") ?? "localhost:6379";
+
         _host = new HostBuilder().ConfigureWebHost(webBuilder =>
         {
             webBuilder
-                //  .UseStartup<Startup>()
-                .UseTestServer();
+                .UseTestServer()
+                .ConfigureServices(services =>
+                {
+                    services.AddGrpc();
+                    services.AddSingleton<ICartStore>(_ =>
+                    {
+                        var store = new ValkeyCartStore(NullLogger<ValkeyCartStore>.Instance, valkeyAddress);
+                        store.Initialize();
+                        return store;
+                    });
+                    services.AddSingleton(sp =>
+                        new cart.services.CartService(
+                            sp.GetRequiredService<ICartStore>(),
+                            new ValkeyCartStore(NullLogger<ValkeyCartStore>.Instance, "badhost:1234"),
+                            Api.Instance.GetClient()));
+                })
+                .Configure(app =>
+                {
+                    app.UseRouting();
+                    app.UseEndpoints(endpoints => endpoints.MapGrpcService<cart.services.CartService>());
+                });
         });
     }
 
-    [Fact(Skip = "See https://github.com/open-telemetry/opentelemetry-demo/pull/746#discussion_r1107931240")]
-    public async Task GetItem_NoAddItemBefore_EmptyCartReturned()
+    private async Task<(IHost server, CartServiceClient client)> StartAsync()
     {
-        // Setup test server and client
-        using var server = await _host.StartAsync();
+        var server = await _host.StartAsync();
         var httpClient = server.GetTestClient();
-
-        string userId = Guid.NewGuid().ToString();
-
-        // Create a GRPC communication channel between the client and the server
         var channel = GrpcChannel.ForAddress(httpClient.BaseAddress, new GrpcChannelOptions
         {
             HttpClient = httpClient
         });
+        return (server, new CartServiceClient(channel));
+    }
 
-        var cartClient = new CartServiceClient(channel);
+    [Fact]
+    public async Task GetItem_NoAddItemBefore_EmptyCartReturned()
+    {
+        var (server, client) = await StartAsync();
+        using var _ = server;
 
-        var request = new GetCartRequest
-        {
-            UserId = userId,
-        };
+        string userId = Guid.NewGuid().ToString();
 
-        var cart = await cartClient.GetCartAsync(request);
+        var cart = await client.GetCartAsync(new GetCartRequest { UserId = userId });
         Assert.NotNull(cart);
 
         // All grpc objects implement IEquitable, so we can compare equality with by-value semantics
         Assert.Equal(new Cart(), cart);
     }
 
-    [Fact(Skip = "See https://github.com/open-telemetry/opentelemetry-demo/pull/746#discussion_r1107931240")]
+    [Fact]
     public async Task AddItem_ItemExists_Updated()
     {
-        // Setup test server and client
-        using var server = await _host.StartAsync();
-        var httpClient = server.GetTestClient();
+        var (server, client) = await StartAsync();
+        using var _ = server;
 
         string userId = Guid.NewGuid().ToString();
-
-        // Create a GRPC communication channel between the client and the server
-        var channel = GrpcChannel.ForAddress(httpClient.BaseAddress, new GrpcChannelOptions
-        {
-            HttpClient = httpClient
-        });
-
-        var client = new CartServiceClient(channel);
         var request = new AddItemRequest
         {
             UserId = userId,
@@ -86,11 +103,7 @@ public async Task AddItem_ItemExists_Updated()
         // Second add of existing product - quantity should be updated
         await client.AddItemAsync(request);
 
-        var getCartRequest = new GetCartRequest
-        {
-            UserId = userId
-        };
-        var cart = await client.GetCartAsync(getCartRequest);
+        var cart = await client.GetCartAsync(new GetCartRequest { UserId = userId });
         Assert.NotNull(cart);
         Assert.Equal(userId, cart.UserId);
         Assert.Single(cart.Items);
@@ -100,24 +113,13 @@ public async Task AddItem_ItemExists_Updated()
         await client.EmptyCartAsync(new EmptyCartRequest { UserId = userId });
     }
 
-    [Fact(Skip = "See https://github.com/open-telemetry/opentelemetry-demo/pull/746#discussion_r1107931240")]
+    [Fact]
     public async Task AddItem_New_Inserted()
     {
-        // Setup test server and client
-        using var server = await _host.StartAsync();
-        var httpClient = server.GetTestClient();
+        var (server, client) = await StartAsync();
+        using var _ = server;
 
         string userId = Guid.NewGuid().ToString();
-
-        // Create a GRPC communication channel between the client and the server
-        var channel = GrpcChannel.ForAddress(httpClient.BaseAddress, new GrpcChannelOptions
-        {
-            HttpClient = httpClient
-        });
-
-        // Create a proxy object to work with the server
-        var client = new CartServiceClient(channel);
-
         var request = new AddItemRequest
         {
             UserId = userId,
@@ -130,10 +132,7 @@ public async Task AddItem_New_Inserted()
 
         await client.AddItemAsync(request);
 
-        var getCartRequest = new GetCartRequest
-        {
-            UserId = userId
-        };
+        var getCartRequest = new GetCartRequest { UserId = userId };
         var cart = await client.GetCartAsync(getCartRequest);
         Assert.NotNull(cart);
         Assert.Equal(userId, cart.UserId);
@@ -143,4 +142,62 @@ public async Task AddItem_New_Inserted()
         cart = await client.GetCartAsync(getCartRequest);
         Assert.Empty(cart.Items);
     }
+
+    [Fact]
+    public async Task AddItem_ConcurrentCallsSameProduct_NoLostUpdates()
+    {
+        var (server, client) = await StartAsync();
+        using var _ = server;
+
+        string userId = Guid.NewGuid().ToString();
+        const int concurrentCalls = 20;
+
+        var tasks = Enumerable.Range(0, concurrentCalls).Select(_ => client.AddItemAsync(new AddItemRequest
+        {
+            UserId = userId,
+            Item = new CartItem { ProductId = "race-product", Quantity = 1 }
+        }).ResponseAsync);
+
+        await Task.WhenAll(tasks);
+
+        var cart = await client.GetCartAsync(new GetCartRequest { UserId = userId });
+        Assert.Single(cart.Items);
+        Assert.Equal(concurrentCalls, cart.Items[0].Quantity);
+
+        await client.EmptyCartAsync(new EmptyCartRequest { UserId = userId });
+    }
+
+    [Fact]
+    public async Task AddItem_NegativeDeltaDuringCheckoutRace_OnlyRemovesOrderedQuantity()
+    {
+        var (server, client) = await StartAsync();
+        using var _ = server;
+
+        string userId = Guid.NewGuid().ToString();
+
+        await client.AddItemAsync(new AddItemRequest
+        {
+            UserId = userId,
+            Item = new CartItem { ProductId = "ordered-product", Quantity = 2 }
+        });
+
+        await client.AddItemAsync(new AddItemRequest
+        {
+            UserId = userId,
+            Item = new CartItem { ProductId = "concurrently-added-product", Quantity = 1 }
+        });
+
+        await client.AddItemAsync(new AddItemRequest
+        {
+            UserId = userId,
+            Item = new CartItem { ProductId = "ordered-product", Quantity = -2 }
+        });
+
+        var cart = await client.GetCartAsync(new GetCartRequest { UserId = userId });
+        var remaining = Assert.Single(cart.Items);
+        Assert.Equal("concurrently-added-product", remaining.ProductId);
+        Assert.Equal(1, remaining.Quantity);
+
+        await client.EmptyCartAsync(new EmptyCartRequest { UserId = userId });
+    }
 }