dgraph-io/dgraph · error

send data downstream(%d): %w

Error message

send data downstream(%d): %w

What it means

For each normal data chunk, pipeTwoStream forwards the packet to the leader, waits for the leader's ack via out.Recv(), and only then acks upstream. If the downstream send of a data packet fails, the ack chain is broken and the relay aborts with this wrapped error naming the group.

Source

Thrown at worker/import.go:444

				if errors.Is(err, io.EOF) {
					return fmt.Errorf("downstream(%d) closed before Finish=true", groupId)
				}
				if err != nil {
					return fmt.Errorf("recv final downstream(%d): %w", groupId, err)
				}
				if err := in.Send(resp); err != nil {
					return fmt.Errorf("relay final upstream: %w", err)
				}
				if resp.Finish {
					glog.Infof("[import] [forward %d -> %d] finish", currentGroup, groupId)
					return nil
				}
			}
		}

		// Normal data chunk: send -> wait ack -> send upstream ack.
		if err := out.Send(&api.StreamExtSnapshotRequest{Pkt: req.Pkt}); err != nil {
			return fmt.Errorf("send data downstream(%d): %w", groupId, err)
		}
		if _, err := out.Recv(); err != nil {
			return fmt.Errorf("ack data downstream(%d): %w", groupId, err)
		}
		if err := in.Send(&api.StreamExtSnapshotResponse{}); err != nil {
			return fmt.Errorf("send ack upstream: %w", err)
		}

	}
}

func (w *grpcWorker) UpdateExtSnapshotStreamingState(ctx context.Context,
	req *api.UpdateExtSnapshotStreamingStateRequest) (*pb.Status, error) {
	if req == nil {
		return nil, errors.New("UpdateExtSnapshotStreamingStateRequest must not be nil")
	}

	if req.Start && req.Finish {

View on GitHub (pinned to 759e242be6)

Solutions

1. Check the leader's logs for the failure that broke the stream, then retry the import
2. Verify direct, stable connectivity (and keepalives) between the two alphas
3. Ensure the leader has disk/memory headroom to keep acking under load
4. Exclude internal gRPC ports from proxy idle timeouts

// stable path + leader health check before starting bulk transfer
pingLeader(leaderAddr); checkKeepaliveConfig(grpcDialOpts)

func isDownstreamBreak(err error) bool {
    c := status.Code(err)
    return errors.Is(err, io.EOF) || c == codes.Unavailable || c == codes.Canceled
}

if err != nil && strings.Contains(err.Error(), "send data downstream") {
    // mid-transfer break; check leader logs/network, then re-run import
}

When it happens

Trigger: out.Send(&api.StreamExtSnapshotRequest{Pkt: req.Pkt}) errors mid-transfer: connection to the leader dropped, the leader cancelled or errored its stream, or gRPC flow control stalled until the context deadline hit.

Common situations: Leader crash mid-transfer; network instability between alphas during a large snapshot; connection-killing middleboxes during long quiet ack periods; leader-side backpressure exceeding client keepalive timeouts.

Related errors

• recv upstream(%d): %w
• send done downstream(%d): %w
• relay final upstream: %w
• while calling MovePredicate
• failed to establish stream with leader: %v

AI-assisted analysis of dgraph-io/dgraph@759e242be6 (2026-09-01). Data as JSON: /api/errors/208c188b536e8b1c. Report an issue: GitHub.