{"record":{"id":"67d26736d4f34ca6","repo":"vitessio/vitess","slug":"vstreamer-is-not-open","errorCode":null,"errorMessage":"VStreamer is not open","messagePattern":"VStreamer is not open","errorType":"exception","errorClass":null,"httpStatus":null,"severity":"error","filePath":"go/vt/vttablet/tabletserver/vstreamer/engine.go","lineNumber":255,"sourceCode":"// Stream starts a new stream.\n// This streams events from the binary logs\nfunc (vse *Engine) Stream(ctx context.Context, startPos string, tablePKs []*binlogdatapb.TableLastPK,\n\tfilter *binlogdatapb.Filter, throttlerApp throttlerapp.Name,\n\tsend func([]*binlogdatapb.VEvent) error, options *binlogdatapb.VStreamOptions,\n) error {\n\tif err := vse.validateBinlogRowImage(ctx, vse.se.GetDBConnector()); err != nil {\n\t\treturn err\n\t}\n\n\t// Ensure vschema is initialized and the watcher is started.\n\t// Starting of the watcher has to be delayed till the first call to Stream\n\t// because this overhead should be incurred only if someone uses this feature.\n\tvse.watcherOnce.Do(vse.setWatch)\n\n\t// Create stream and add it to the map.\n\tstreamer, idx, err := func() (*uvstreamer, int, error) {\n\t\tif atomic.LoadInt32(&vse.isOpen) == 0 {\n\t\t\treturn nil, 0, errors.New(\"VStreamer is not open\")\n\t\t}\n\t\tvse.mu.Lock()\n\t\tdefer vse.mu.Unlock()\n\t\tstreamer := newUVStreamer(ctx, vse, vse.env.Config().DB.FilteredWithDB(), vse.se, startPos, tablePKs,\n\t\t\tfilter, vse.lvschema, throttlerApp, send, options)\n\t\tidx := vse.streamIdx\n\t\tvse.streamers[idx] = streamer\n\t\tvse.streamIdx++\n\t\t// Now that we've added the stream, increment wg.\n\t\t// This must be done before releasing the lock.\n\t\tvse.wg.Add(1)\n\t\treturn streamer, idx, nil\n\t}()\n\tif err != nil {\n\t\treturn err\n\t}\n\n\t// Remove stream from map and decrement wg when it ends.","sourceCodeStart":237,"sourceCodeEnd":273,"githubUrl":"https://github.com/vitessio/vitess/blob/01a25a7d176f94613b8d59d799f438380a8760e4/go/vt/vttablet/tabletserver/vstreamer/engine.go#L237-L273","documentation":"The VStreamer Engine rejects a new UVStream request when its atomic isOpen flag is 0, i.e. the VStreamer engine has not been opened (Engine.Open) or was already closed. Streams can only be created while the engine is open, so the request fails with \"VStreamer is not open\".","triggerScenarios":"Calling Engine.UVStream (via VStream API / VReplication workflow) before tabletserver finishes opening the VStreamer engine, or after Close during shutdown.","commonSituations":"VReplication workflow start racing vttablet startup or shutdown; move-tables/resume operations hitting a tablet that is being restarted; unit tests using an unopened Engine.","solutions":["Wait for the tablet to fully start (VStreamer engine opened) and retry the VStream request.","Check vttablet logs/errors during startup — the engine may have failed to open due to topo or DB issues.","Point the VReplication workflow at a tablet that is up and serving, not one mid-restart.","In tests, call vse.Open() before streaming and defer vse.Close()."],"exampleFix":"// before\nerr := vstreamerEngine.UVStream(ctx, startPos, nil, filter, app, send, options) // engine closed\n// after\nif err := vstreamerEngine.Open(ctx); err != nil { return err }\nerr := vstreamerEngine.UVStream(ctx, startPos, nil, filter, app, send, options)","handlingStrategy":"retry","validationCode":"if !vse.IsOpen() { return errors.New(\"VStreamer engine not open; retry after tablet startup\") }","typeGuard":"func (vse *Engine) IsOpen() bool { return atomic.LoadInt32(&vse.isOpen) > 0 }","tryCatchPattern":"if err := engine.UVStream(...); err != nil && strings.Contains(err.Error(), \"VStreamer is not open\") {\n  // retry with backoff after tablet readiness\n}","preventionTips":["Gate VReplication workflow starts on tablet SERVING status.","Add backoff retries around VStream calls.","Check tablet logs for Engine.Open failures after restarts."],"tags":["vitess","vstreamer","vreplication","lifecycle"],"backgroundTag":"service-not-initialized","analyzedSha":"01a25a7d176f94613b8d59d799f438380a8760e4","analyzedAt":"2026-09-01T17:28:30.605Z","contentChangedAt":null,"schemaVersion":2},"datasetVersion":"2026-09-08T10:18:20.063Z"}