Pipeline the calls involved in cache set

This makes these calls both pipelined (all executed with a single
network call) and transactional (all must succeed or none will).

cache/cache.go

@@ -218,25 +218,17 @@ func (c *Cache[T]) set(ctx context.Context, key string, value T) error {
 		return err
 	}
 
-	// Remove any explicit nonexistence sentinel
-	err = c.client.Del(ctx, keys.negative).Err()
-	if err != nil {
-		return err
-	}
+	pipe := c.client.TxPipeline()
 
+	// Remove any explicit nonexistence sentinel
+	pipe.Del(ctx, keys.negative)
 	// Update cached value
-	err = c.client.Set(ctx, keys.data, string(data), c.opts.Stale).Err()
-	if err != nil {
-		return err
-	}
-
+	pipe.Set(ctx, keys.data, string(data), c.opts.Stale)
 	// Set freshness sentinel
-	err = c.client.Set(ctx, keys.fresh, 1, c.opts.Fresh).Err()
-	if err != nil {
-		return err
-	}
+	pipe.Set(ctx, keys.fresh, 1, c.opts.Fresh)
 
-	return nil
+	_, err = pipe.Exec(ctx)
+	return err
 }
 
 func (c *Cache[T]) setNegative(ctx context.Context, key string) error {

cache/cache_test.go

@@ -72,9 +72,11 @@ func (m mockWrapper) ExpectCacheFill(key string, value any) {
 	if err != nil {
 		panic(err)
 	}
+	m.ExpectTxPipeline()
 	m.ExpectDel("cache:negative:" + m.name + ":" + key).SetVal(0)
 	m.ExpectSet("cache:data:"+m.name+":"+key, string(data), m.stale).SetVal("OK")
 	m.ExpectSet("cache:fresh:"+m.name+":"+key, 1, m.fresh).SetVal("OK")
+	m.ExpectTxPipelineExec()
 }
 
 func (m mockWrapper) ExpectCacheFillNegative(key string) {