{"record":{"id":"897a8d1cb1be65e7","repo":"dgraph-io/dgraph","slug":"ack-data-downstream-d-w","errorCode":null,"errorMessage":"ack data downstream(%d): %w","messagePattern":"ack data downstream\\((.+?)\\): %w","errorType":"error_code","errorClass":null,"httpStatus":null,"severity":"error","filePath":"worker/import.go","lineNumber":447,"sourceCode":"\t\t\t\tif err != nil {\n\t\t\t\t\treturn fmt.Errorf(\"recv final downstream(%d): %w\", groupId, err)\n\t\t\t\t}\n\t\t\t\tif err := in.Send(resp); err != nil {\n\t\t\t\t\treturn fmt.Errorf(\"relay final upstream: %w\", err)\n\t\t\t\t}\n\t\t\t\tif resp.Finish {\n\t\t\t\t\tglog.Infof(\"[import] [forward %d -> %d] finish\", currentGroup, groupId)\n\t\t\t\t\treturn nil\n\t\t\t\t}\n\t\t\t}\n\t\t}\n\n\t\t// Normal data chunk: send -> wait ack -> send upstream ack.\n\t\tif err := out.Send(&api.StreamExtSnapshotRequest{Pkt: req.Pkt}); err != nil {\n\t\t\treturn fmt.Errorf(\"send data downstream(%d): %w\", groupId, err)\n\t\t}\n\t\tif _, err := out.Recv(); err != nil {\n\t\t\treturn fmt.Errorf(\"ack data downstream(%d): %w\", groupId, err)\n\t\t}\n\t\tif err := in.Send(&api.StreamExtSnapshotResponse{}); err != nil {\n\t\t\treturn fmt.Errorf(\"send ack upstream: %w\", err)\n\t\t}\n\n\t}\n}\n\nfunc (w *grpcWorker) UpdateExtSnapshotStreamingState(ctx context.Context,\n\treq *api.UpdateExtSnapshotStreamingStateRequest) (*pb.Status, error) {\n\tif req == nil {\n\t\treturn nil, errors.New(\"UpdateExtSnapshotStreamingStateRequest must not be nil\")\n\t}\n\n\tif req.Start && req.Finish {\n\t\treturn nil, errors.New(\"UpdateExtSnapshotStreamingStateRequest cannot have both Start and Finish set to true\")\n\t}\n","sourceCodeStart":429,"sourceCodeEnd":465,"githubUrl":"https://github.com/dgraph-io/dgraph/blob/759e242be62c91f8d084da06ad0c8d21256d9c07/worker/import.go#L429-L465","documentation":"During external snapshot streaming, pipeTwoStream relays each data chunk from the upstream (proxy) stream to the downstream group-leader stream. After sending a chunk downstream it waits for the downstream ack via out.Recv(); if that receive fails, the chunk is not acknowledged downstream and the pipe aborts with this wrapped error.","triggerScenarios":"The downstream gRPC StreamExtSnapshot stream breaks (peer alpha restarted, network drop, stream deadline, context cancellation) between out.Send of a data packet and its ack Recv, or the downstream peer returns a non-EOF error on its ack response.","commonSituations":"Follower/leader alpha restarted mid-import; network partition or load balancer idle timeout between alphas; import run canceled or context deadline exceeded; downstream node crashed while applying the snapshot.","solutions":["Check connectivity and health of the downstream alpha that should ack the stream, then retry the snapshot import.","Increase gRPC keepalive/deadline settings so long idles between chunks are not terminated.","Ensure the downstream group leader is up and serving the target group (check membership state).","Retry the whole external snapshot streaming operation; the pipe aborts the stream, so a partial retry is not possible."],"exampleFix":"// before (no resilience around the stream)\nclient.StreamExtSnapshot(ctx)\n// after (bound the stream and retry on failure)\nfor attempt := 0; attempt < 3; attempt++ {\n    ctx, cancel := context.WithTimeout(context.Background(), 30*time.Minute)\n    err := streamSnapshot(ctx)\n    cancel()\n    if err == nil { break }\n    time.Sleep(5 * time.Second)\n}","handlingStrategy":"retry","validationCode":"// before starting the import, check the downstream alpha is reachable\nconn, err := grpc.DialContext(ctx, addr, grpc.WithBlock(), grpc.WithTimeout(10*time.Second))\nif err != nil { return fmt.Errorf(\"downstream %s unreachable: %w\", addr, err) }","typeGuard":null,"tryCatchPattern":"err := streamSnapshot(ctx)\nvar retriable = isTransportErr(err) // match 'ack data downstream' wrapper\nif retriable { backoff-and-retry whole snapshot import }","preventionTips":["Set generous deadlines and gRPC keepalives for long snapshot streams","Verify all alphas are healthy before starting an external snapshot import","Avoid importing across unreliable network links or aggressive LB idle timeouts"],"tags":["grpc","streaming","network","distributed-systems"],"backgroundTag":"grpc-stream-aborted","analyzedSha":"759e242be62c91f8d084da06ad0c8d21256d9c07","analyzedAt":"2026-09-01T14:42:12.034Z","contentChangedAt":null,"schemaVersion":2},"datasetVersion":"2026-09-08T10:18:20.063Z"}