{"record":{"id":"bf08cbe27f3f8cd7","repo":"vitessio/vitess","slug":"row-stream-ended-w","errorCode":null,"errorMessage":"row stream ended: %w","messagePattern":"row stream ended: %w","errorType":"exception","errorClass":null,"httpStatus":null,"severity":"info","filePath":"go/vt/vttablet/tabletserver/vstreamer/rowstreamer.go","lineNumber":355,"sourceCode":"\tbuf.Myprintf(\" order by \", sqlparser.NewIdentifierCS(rs.plan.Table.Name))\n\tprefix = \"\"\n\tfor _, pk := range rs.pkColumns {\n\t\tbuf.Myprintf(\"%s%v\", prefix, sqlparser.NewIdentifierCI(rs.plan.Table.Fields[pk].Name))\n\t\tprefix = \", \"\n\t}\n\treturn buf.String(), nil\n}\n\nfunc (rs *rowStreamer) streamQuery(send func(*binlogdatapb.VStreamRowsResponse) error) error {\n\tthrottleResponseRateLimiter := timer.NewRateLimiter(rowStreamertHeartbeatInterval)\n\tdefer throttleResponseRateLimiter.Stop()\n\n\tvar sendMu sync.Mutex\n\tsafeSend := func(ctx context.Context, r *binlogdatapb.VStreamRowsResponse) error {\n\t\tsendMu.Lock()\n\t\tdefer sendMu.Unlock()\n\t\tif err := ctx.Err(); err != nil {\n\t\t\treturn fmt.Errorf(\"row stream ended: %w\", err)\n\t\t}\n\t\treturn send(r)\n\t}\n\t// Let's wait until MySQL is in good shape to stream rows\n\tif err := rs.vse.waitForMySQL(rs.ctx, rs.cp, rs.plan.Table.Name); err != nil {\n\t\treturn err\n\t}\n\tvar (\n\t\tgtid       string\n\t\trotatedLog bool\n\t\terr        error\n\t)\n\tlog.Info(fmt.Sprintf(\"Streaming rows for query: %s\\n\", rs.sendQuery))\n\tif rs.mode == RowStreamerModeSingleTable {\n\t\tgtid, rotatedLog, err = rs.conn.streamWithSnapshot(rs.ctx, rs.plan.Table.Name, rs.sendQuery)\n\t\tif err != nil {\n\t\t\treturn err\n\t\t}","sourceCodeStart":337,"sourceCodeEnd":373,"githubUrl":"https://github.com/vitessio/vitess/blob/01a25a7d176f94613b8d59d799f438380a8760e4/go/vt/vttablet/tabletserver/vstreamer/rowstreamer.go#L337-L373","documentation":"safeSend wraps context cancellation during a VStreamRows send attempt: if the row streamer's context is already done when it tries to send a response, it returns 'row stream ended' with the context error. It signals orderly termination of the row stream, not a data error.","triggerScenarios":"rs.ctx cancelled/deadline exceeded between events during Stream → streamQuery; the next safeSend call observes ctx.Err() and returns this error.","commonSituations":"VReplication workflow cancelled or completed, tablet graceful shutdown during copy or streaming phase, caller-side timeout on VStreamRows.","solutions":["Read the wrapped cause: Canceled is expected on workflow stop/shutdown; DeadlineExceeded indicates the caller's timeout is too short for the copy volume.","If streams stop unexpectedly, check tablet logs just before the error for shutdown/OOM and verify the workflow wasn't cancelled by another actor.","Re-invoke VStream; it resumes from the lastpk/GTID checkpoint."],"exampleFix":null,"handlingStrategy":"retry","validationCode":null,"typeGuard":null,"tryCatchPattern":"err := rowStream(ctx, ...)\nif err != nil {\n    var ctxErr error\n    if errors.As(err, &ctxErr) && (errors.Is(err, context.Canceled) || errors.Is(err, context.DeadlineExceeded)) {\n        // orderly stop: log at info, resume from checkpoint\n        resumeFromCheckpoint(ctx)\n    }\n}","preventionTips":["Size stream timeouts to table size","Cancel workflows explicitly and expect this error during teardown","Keep gRPC connections alive (no aggressive LB idle timeouts)","Resume from lastpk instead of re-copying"],"tags":["vstreamer","rowstreamer","context-canceled"],"backgroundTag":"context-canceled-during-stream","analyzedSha":"01a25a7d176f94613b8d59d799f438380a8760e4","analyzedAt":"2026-09-01T17:28:30.605Z","contentChangedAt":null,"schemaVersion":2},"datasetVersion":"2026-09-08T10:18:20.063Z"}