{"record":{"id":"64593ee2732ea2d4","repo":"dgraph-io/dgraph","slug":"send-ack-upstream-w","errorCode":null,"errorMessage":"send ack upstream: %w","messagePattern":"send ack upstream: %w","errorType":"error_code","errorClass":null,"httpStatus":null,"severity":"error","filePath":"worker/import.go","lineNumber":450,"sourceCode":"\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\n\tglog.Infof(\"[import] Applying import mode proposal: %+v\", req)\n\terr := groups().Node.proposeAndWait(ctx, &pb.Proposal{ExtSnapshotState: req})\n","sourceCodeStart":432,"sourceCodeEnd":468,"githubUrl":"https://github.com/dgraph-io/dgraph/blob/759e242be62c91f8d084da06ad0c8d21256d9c07/worker/import.go#L432-L468","documentation":"After a data chunk is acked by the downstream peer, pipeTwoStream sends the ack back upstream on the proxy's stream (in.Send). If that send fails, the upstream caller never receives the ack and the relay aborts with this wrapped error.","triggerScenarios":"The upstream gRPC stream from the calling alpha has been closed, reset, or its context canceled while pipeTwoStream tries to send the StreamExtSnapshotResponse ack — e.g. client disconnected, deadline exceeded, or transport broken.","commonSituations":"Proxy alpha (or the dgraph live/backup client driving it) dropped its connection mid-import; client-side timeout shorter than streaming duration; client canceled the import.","solutions":["Check the client that opened the stream: it likely disconnected or timed out; increase its deadline and retry.","Verify network stability between this alpha and the upstream caller.","Retry the snapshot streaming operation end-to-end after fixing the connection.","Make sure no middleware/proxy (LB) kills long-lived gRPC streams; enable keepalives."],"exampleFix":"// before: client ctx with short timeout\nctx, cancel := context.WithTimeout(ctx, 30*time.Second)\n// after: generous timeout for bulk streaming\nctx, cancel := context.WithTimeout(ctx, 2*time.Hour)","handlingStrategy":"retry","validationCode":null,"typeGuard":null,"tryCatchPattern":"if err := runImport(ctx); err != nil {\n    if strings.Contains(err.Error(), \"send ack upstream\") || status.Code(errors.Unwrap(err)) == codes.Unavailable {\n        // client disconnected: fix client and retry\n    }\n}","preventionTips":["Keep the importing client connected for the whole stream (no aggressive client timeouts)","Enable gRPC keepalive on the client","Do not cancel the client context mid-import; abort cleanly via the API instead"],"tags":["grpc","streaming","network"],"backgroundTag":"grpc-stream-aborted","analyzedSha":"759e242be62c91f8d084da06ad0c8d21256d9c07","analyzedAt":"2026-09-01T14:42:12.034Z","contentChangedAt":null,"schemaVersion":2},"datasetVersion":"2026-09-08T10:18:20.063Z"}