wip: idempotency

src/gen/io/defang/v1/fabric_connect.d.ts

@@ -53,6 +53,7 @@ export declare const FabricController: {
       readonly I: typeof Empty,
       readonly O: typeof Empty,
       readonly kind: MethodKind.Unary,
+      readonly idempotency: MethodIdempotency.Idempotent,
     },
     /**
      * @generated from rpc io.defang.v1.FabricController.Tail
@@ -62,6 +63,7 @@ export declare const FabricController: {
       readonly I: typeof TailRequest,
       readonly O: typeof TailResponse,
       readonly kind: MethodKind.ServerStreaming,
+      readonly idempotency: MethodIdempotency.NoSideEffects,
     },
     /**
      * @generated from rpc io.defang.v1.FabricController.Update
@@ -119,6 +121,7 @@ export declare const FabricController: {
       readonly I: typeof SubscribeRequest,
       readonly O: typeof SubscribeResponse,
       readonly kind: MethodKind.ServerStreaming,
+      readonly idempotency: MethodIdempotency.NoSideEffects,
     },
     /**
      * rpc Promote(google.protobuf.Empty) returns (google.protobuf.Empty);

src/gen/io/defang/v1/fabric_connect.js

@@ -53,6 +53,7 @@ export const FabricController = {
       I: Empty,
       O: Empty,
       kind: MethodKind.Unary,
+      idempotency: MethodIdempotency.Idempotent,
     },
     /**
      * @generated from rpc io.defang.v1.FabricController.Tail
@@ -62,6 +63,7 @@ export const FabricController = {
       I: TailRequest,
       O: TailResponse,
       kind: MethodKind.ServerStreaming,
+      idempotency: MethodIdempotency.NoSideEffects,
     },
     /**
      * @generated from rpc io.defang.v1.FabricController.Update
@@ -119,6 +121,7 @@ export const FabricController = {
       I: SubscribeRequest,
       O: SubscribeResponse,
       kind: MethodKind.ServerStreaming,
+      idempotency: MethodIdempotency.NoSideEffects,
     },
     /**
      * rpc Promote(google.protobuf.Empty) returns (google.protobuf.Empty);

src/gen/io/defang/v1/fabric_pb.d.ts

@@ -203,9 +203,9 @@ export declare class DeployRequest extends Message<DeployRequest> {
   project: string;
 
   /**
-   * @generated from field: google.protobuf.Struct compose = 3;
+   * @generated from field: string compose_yaml = 3;
    */
-  compose?: Struct;
+  composeYaml: string;
 
   constructor(data?: PartialMessage<DeployRequest>);
 

src/gen/io/defang/v1/fabric_pb.js

@@ -90,7 +90,7 @@ export const DeployRequest = /*@__PURE__*/ proto3.makeMessageType(
   () => [
     { no: 1, name: "services", kind: "message", T: Service, repeated: true },
     { no: 2, name: "project", kind: "scalar", T: 9 /* ScalarType.STRING */ },
-    { no: 3, name: "compose", kind: "message", T: Struct },
+    { no: 3, name: "compose_yaml", kind: "scalar", T: 9 /* ScalarType.STRING */ },
   ],
 );
 

src/pkg/cli/client/retrier.go

@@ -13,7 +13,7 @@ type Retrier struct{}
 func (Retrier) WrapUnary(next connect.UnaryFunc) connect.UnaryFunc {
 	return func(ctx context.Context, req connect.AnyRequest) (connect.AnyResponse, error) {
 		res, err := next(ctx, req)
-		if connect.CodeOf(err) == connect.CodeUnavailable {
+		if connect.CodeOf(err) == connect.CodeUnavailable || req.Spec().IdempotencyLevel == connect.IdempotencyNoSideEffects {
 			// Retry once after a 1 second sleep
 			pkg.SleepWithContext(ctx, 1*time.Second)
 			res, err = next(ctx, req)

src/pkg/cli/client/retrier_test.go

@@ -5,6 +5,7 @@ import (
 	"errors"
 	"net/http/httptest"
 	"testing"
+	"time"
 
 	defangv1 "github.com/DefangLabs/defang/src/protos/io/defang/v1"
 	"github.com/DefangLabs/defang/src/protos/io/defang/v1/defangv1connect"
@@ -13,18 +14,28 @@ import (
 
 type grpcMockHandler struct {
 	defangv1connect.UnimplementedFabricControllerHandler
-	tries int
+	getTries    int
+	deployTries int
+	tailTries   int
+}
+
+func (g *grpcMockHandler) Get(context.Context, *connect.Request[defangv1.ServiceID]) (*connect.Response[defangv1.ServiceInfo], error) {
+	println(time.Now().Format(time.RFC3339Nano), "Get")
+	g.getTries++
+	return nil, connect.NewError(connect.CodeUnavailable, errors.New("unavailable"))
 }
 
 func (g *grpcMockHandler) Deploy(context.Context, *connect.Request[defangv1.DeployRequest]) (*connect.Response[defangv1.DeployResponse], error) {
-	g.tries++
+	println(time.Now().Format(time.RFC3339Nano), "Deploy")
+	g.deployTries++
 	return nil, connect.NewError(connect.CodeUnavailable, errors.New("unavailable"))
 }
 
-// func (g *grpcMockHandler) Tail(ctx context.Context, r *connect.Request[defangv1.TailRequest], s *connect.ServerStream[defangv1.TailResponse]) error {
-// 	g.tries++
-// 	return connect.NewError(connect.CodeUnavailable, errors.New("unavailable"))
-// }
+func (g *grpcMockHandler) Tail(ctx context.Context, r *connect.Request[defangv1.TailRequest], s *connect.ServerStream[defangv1.TailResponse]) error {
+	println(time.Now().Format(time.RFC3339Nano), "Tail")
+	g.tailTries++
+	return connect.NewError(connect.CodeUnavailable, errors.New("unavailable"))
+}
 
 func TestRetrier(t *testing.T) {
 	fabricServer := &grpcMockHandler{}
@@ -35,11 +46,41 @@ func TestRetrier(t *testing.T) {
 
 	fabricClient := defangv1connect.NewFabricControllerClient(server.Client(), server.URL, connect.WithGRPC(), connect.WithInterceptors(Retrier{}))
 
-	_, err := fabricClient.Deploy(context.Background(), connect.NewRequest(&defangv1.DeployRequest{}))
-	if err == nil {
-		t.Fatal("expected error")
-	}
-	if fabricServer.tries != 2 {
-		t.Fatalf("expected 2 tries, got %d", fabricServer.tries)
-	}
+	t.Run("Unary idempotent", func(t *testing.T) {
+		_, err := fabricClient.Get(context.Background(), connect.NewRequest(&defangv1.ServiceID{}))
+		if err == nil {
+			t.Fatal("expected error")
+		}
+		if fabricServer.getTries != 2 {
+			t.Fatalf("expected 2 tries, got %d", fabricServer.getTries)
+		}
+	})
+
+	t.Run("Unary", func(t *testing.T) {
+		_, err := fabricClient.Deploy(context.Background(), connect.NewRequest(&defangv1.DeployRequest{}))
+		if err == nil {
+			t.Fatal("expected error")
+		}
+		if fabricServer.deployTries != 2 {
+			t.Fatalf("expected 2 tries, got %d", fabricServer.deployTries)
+		}
+	})
+
+	t.Run("Streaming", func(t *testing.T) {
+		ss, err := fabricClient.Tail(context.Background(), connect.NewRequest(&defangv1.TailRequest{}))
+		if err != nil {
+			t.Fatal(err)
+		}
+		defer ss.Close()
+		if ss.Receive() == true {
+			t.Error("expected false")
+		}
+		if ss.Err() == nil {
+			t.Errorf("expected error")
+		}
+		// TODO: implement retries for streaming calls
+		if fabricServer.tailTries != 1 {
+			t.Fatalf("expected 1 tries, got %d", fabricServer.tailTries)
+		}
+	})
 }