{"record":{"id":"c2ccd1b61875fc59","repo":"dapr/dapr","slug":"duplicate-initial-request-received","errorCode":null,"errorMessage":"duplicate initial request received","messagePattern":"duplicate initial request received","errorType":"exception","errorClass":null,"httpStatus":null,"severity":"error","filePath":"pkg/runtime/pubsub/streamer/streamer.go","lineNumber":167,"sourceCode":"\t\tresp, err := stream.Recv()\n\n\t\tstat, ok := status.FromError(err)\n\n\t\tif (ok && stat.Code() == codes.Canceled) ||\n\t\t\terrors.Is(err, context.Canceled) ||\n\t\t\terrors.Is(err, io.EOF) {\n\t\t\tlog.Infof(\"Unsubscribed from pubsub '%s' topic '%s'\", req.GetPubsubName(), req.GetTopic())\n\t\t\treturn err\n\t\t}\n\n\t\tif err != nil {\n\t\t\tlog.Errorf(\"Error receiving message from client stream: %s\", err)\n\t\t\treturn err\n\t\t}\n\n\t\teventResp := resp.GetEventProcessed()\n\t\tif eventResp == nil {\n\t\t\treturn errors.New(\"duplicate initial request received\")\n\t\t}\n\n\t\tconn.notifyPublishResponse(eventResp)\n\t}\n}\n\n// TODO: @joshvanl: move diagnostics.\nfunc (s *streamer) Publish(ctx context.Context, msg *rtpubsub.SubscribedMessage) (*rtv1pb.SubscribeTopicEventsRequestProcessedAlpha1, error) {\n\ts.lock.RLock()\n\tkey := s.StreamerKey(msg.PubSub, msg.Topic)\n\tconnection, ok := s.subscribers[key][msg.SubscriberID]\n\ts.lock.RUnlock()\n\n\tif !ok {\n\t\treturn nil, fmt.Errorf(\"no streamer subscribed to pubsub %q topic %q\", msg.PubSub, msg.Topic)\n\t}\n\n\tif connection.closed.Load() {","sourceCodeStart":149,"sourceCodeEnd":185,"githubUrl":"https://github.com/dapr/dapr/blob/74ad41702745709bb15fe2114ff693b8c59bc3cc/pkg/runtime/pubsub/streamer/streamer.go#L149-L185","documentation":"Returned by the server-side topic streamer recvLoop when a client already sent its initial SubscribeTopicEventsRequestInitialAlpha1 and then sends another request whose message is not an EventProcessed acknowledgement. The protocol is one initial message followed by event-processed confirmations; anything else (a repeated initial request, or an unknown message variant) fails the stream.","triggerScenarios":"Calling stream.Send with a second initial request on the same SubscribeTopicEvents stream; a client built against an older/newer protocol version that emits a different message sequence; buggy client code looping the initial request.","commonSituations":"Custom client SDKs re-using a stream after reconnect instead of opening a new one; retry logic that re-sends the initial handshake on the same stream; proto version skew between client and sidecar.","solutions":["Send the initial request exactly once per stream; open a new stream (new SubscribeTopicEvents call) on reconnect.","Match the client proto/surface to the sidecar's runtime version (regenerate from the dapr proto used by the deployed runtime).","After receiving an error on the stream, always tear the stream down and resubscribe from scratch rather than continuing.","Check client logs for 'duplicate initial request' and fix the send loop that emits it."],"exampleFix":"// before: re-sending initial on the same stream\nfor {\n    if err := stream.Send(initialReq); err != nil { // second iteration breaks protocol\n        return err\n    }\n}\n\n// after: send once, then only acks\nif err := stream.Send(initialReq); err != nil {\n    return err\n}\nfor ev := range events {\n    stream.Send(&rtv1pb.SubscribeTopicEventsRequestProcessedAlpha1{ ... })\n}","handlingStrategy":"validation","validationCode":"// Client-side: enforce one initial request per stream\ninitialSent := false\nfor {\n    msg := buildNext()\n    if msg.GetInitial() != nil {\n        if initialSent {\n            return errors.New(\"refusing duplicate initial request\")\n        }\n        initialSent = true\n    }\n    if err := stream.Send(msg); err != nil {\n        return err\n    }\n}","typeGuard":null,"tryCatchPattern":"err := stream.RecvLoop()\nif err != nil && strings.Contains(err.Error(), \"duplicate initial request received\") {\n    // protocol violation from our client: reconnect with a fresh stream\n    stream, err = reopenSubscription(ctx)\n}","preventionTips":["On any stream error, close and re-dial the subscription stream.","Pin client protos to the runtime version you deploy.","Track handshake state explicitly in client SDKs."],"tags":["pubsub","streaming","grpc","protocol","client-protocol","go"],"backgroundTag":null,"analyzedSha":"74ad41702745709bb15fe2114ff693b8c59bc3cc","analyzedAt":"2026-08-16T04:22:26.543Z","schemaVersion":2},"datasetVersion":"2026-08-16T08:17:34.114Z"}