Method TestRetryTransparentWhenCommitted

test/retry_test.go:771–851

(t *testing.T)

}

func (s) TestRetryTransparentWhenCommitted(t *testing.T) {
	// With MaxConcurrentStreams=1:
	//
	// 1. Create stream 1 that is retriable.
	// 2. Stream 1 is created and fails with a retriable code.
	// 3. Create dummy stream 2, blocking indefinitely.
	// 4. Stream 1 retries (and blocks until stream 2 finishes)
	// 5. Stream 1 is canceled manually.
	//
	// If there is no bug, the stream is done and errors with CANCELED.  With a bug:
	//
	// 6. Stream 1 has a nil stream (attempt.s).  Operations like CloseSend will panic.

	first := grpcsync.NewEvent()
	ss := &stubserver.StubServer{
		FullDuplexCallF: func(stream testgrpc.TestService_FullDuplexCallServer) error {
			// signal?
			if !first.HasFired() {
				first.Fire()
				t.Log("returned first error")
				return status.Error(codes.AlreadyExists, "first attempt fails and is retriable")
			}
			t.Log("blocking")
			<-stream.Context().Done()
			return stream.Context().Err()
		},
	}

	if err := ss.Start([]grpc.ServerOption{grpc.MaxConcurrentStreams(1)},
		grpc.WithDefaultServiceConfig(`{
    "methodConfig": [{
      "name": [{"service": "grpc.testing.TestService"}],
      "waitForReady": true,
      "retryPolicy": {
        "MaxAttempts": 2,
        "InitialBackoff": ".1s",
        "MaxBackoff": ".1s",
        "BackoffMultiplier": 1.0,
        "RetryableStatusCodes": [ "ALREADY_EXISTS" ]
      }
    }]}`)); err != nil {
		t.Fatalf("Error starting endpoint server: %v", err)
	}
	defer ss.Stop()

	ctx1, cancel1 := context.WithTimeout(context.Background(), defaultTestTimeout)
	defer cancel1()
	ctx2, cancel2 := context.WithTimeout(context.Background(), defaultTestTimeout)
	defer cancel2()

	stream1, err := ss.Client.FullDuplexCall(ctx1)
	if err != nil {
		t.Fatalf("Error creating stream 1: %v", err)
	}

	// Create dummy stream to block indefinitely.
	_, err = ss.Client.FullDuplexCall(ctx2)
	if err != nil {