dapr/dapr · error

duplicate initial request received

Error message

duplicate initial request received

What it means

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.

Source

Thrown at pkg/runtime/pubsub/streamer/streamer.go:167

		resp, err := stream.Recv()

		stat, ok := status.FromError(err)

		if (ok && stat.Code() == codes.Canceled) ||
			errors.Is(err, context.Canceled) ||
			errors.Is(err, io.EOF) {
			log.Infof("Unsubscribed from pubsub '%s' topic '%s'", req.GetPubsubName(), req.GetTopic())
			return err
		}

		if err != nil {
			log.Errorf("Error receiving message from client stream: %s", err)
			return err
		}

		eventResp := resp.GetEventProcessed()
		if eventResp == nil {
			return errors.New("duplicate initial request received")
		}

		conn.notifyPublishResponse(eventResp)
	}
}

// TODO: @joshvanl: move diagnostics.
func (s *streamer) Publish(ctx context.Context, msg *rtpubsub.SubscribedMessage) (*rtv1pb.SubscribeTopicEventsRequestProcessedAlpha1, error) {
	s.lock.RLock()
	key := s.StreamerKey(msg.PubSub, msg.Topic)
	connection, ok := s.subscribers[key][msg.SubscriberID]
	s.lock.RUnlock()

	if !ok {
		return nil, fmt.Errorf("no streamer subscribed to pubsub %q topic %q", msg.PubSub, msg.Topic)
	}

	if connection.closed.Load() {

View on GitHub (pinned to 74ad417027)

Solutions

1. Send the initial request exactly once per stream; open a new stream (new SubscribeTopicEvents call) on reconnect.
2. Match the client proto/surface to the sidecar's runtime version (regenerate from the dapr proto used by the deployed runtime).
3. After receiving an error on the stream, always tear the stream down and resubscribe from scratch rather than continuing.
4. Check client logs for 'duplicate initial request' and fix the send loop that emits it.

Example fix

// before: re-sending initial on the same stream
for {
    if err := stream.Send(initialReq); err != nil { // second iteration breaks protocol
        return err
    }
}

// after: send once, then only acks
if err := stream.Send(initialReq); err != nil {
    return err
}
for ev := range events {
    stream.Send(&rtv1pb.SubscribeTopicEventsRequestProcessedAlpha1{ ... })
}

// Client-side: enforce one initial request per stream
initialSent := false
for {
    msg := buildNext()
    if msg.GetInitial() != nil {
        if initialSent {
            return errors.New("refusing duplicate initial request")
        }
        initialSent = true
    }
    if err := stream.Send(msg); err != nil {
        return err
    }
}

err := stream.RecvLoop()
if err != nil && strings.Contains(err.Error(), "duplicate initial request received") {
    // protocol violation from our client: reconnect with a fresh stream
    stream, err = reopenSubscription(ctx)
}

When it happens

Trigger: 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.

Common situations: 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.

Related errors

• response metadata found in non-leading chunk
• unable to subscribe: %w
• response does not contain the required fields in the leading
• few message(s) have failed during bulk subscribe operation
• unable to read from binding: %w

AI-assisted analysis of dapr/dapr@74ad417027 (2026-08-16). Data as JSON: /api/errors/c2ccd1b61875fc59. Report an issue: GitHub.