{"record":{"id":"61b21ae177f8bf2a","repo":"apache/beam","slug":"decoderestream-opened-twice","errorCode":null,"errorMessage":"decodeReStream opened twice","messagePattern":"decodeReStream opened twice","errorType":"exception","errorClass":null,"httpStatus":null,"severity":"error","filePath":"sdks/go/pkg/beam/core/runtime/exec/fullvalue.go","lineNumber":208,"sourceCode":"\t\t}\n\t\tret = append(ret, *elm)\n\t}\n}\n\n// singleUseReStream is a decode on demand ReStream.\n// Can only produce a single Stream because it consumes the reader.\n// Must not be used for streams that might be re-iterated, causing Open\n// to be called twice.\ntype singleUseReStream struct {\n\tr    io.Reader\n\td    ElementDecoder\n\tsize int // The number of elements in this stream.\n}\n\n// Open returns the Stream from the start of the in-memory reader. Returns error if called twice.\nfunc (n *singleUseReStream) Open() (Stream, error) {\n\tif n.r == nil {\n\t\treturn nil, errors.New(\"decodeReStream opened twice\")\n\t}\n\tret := &decodeStream{r: n.r, d: n.d, size: n.size}\n\tn.r = nil\n\tn.d = nil\n\treturn ret, nil\n}\n\n// decodeStream is a decode on demand Stream, that decodes size elements from the provided\n// io.Reader.\ntype decodeStream struct {\n\tr          io.Reader\n\td          ElementDecoder\n\tnext, size int\n\tret        FullValue\n}\n\n// Close causes subsequent calls to Read to return io.EOF, and drains the remaining element count\n// from the reader.","sourceCodeStart":190,"sourceCodeEnd":226,"githubUrl":"https://github.com/apache/beam/blob/12126d8942aaf848030c478b4c6a28c6af861c66/sdks/go/pkg/beam/core/runtime/exec/fullvalue.go#L190-L226","documentation":"singleUseReStream is a one-shot in-memory ReStream: Open() hands out the underlying reader exactly once and nils it out. A second Open() call finds n.r == nil and returns this error, because the stream cannot be replayed from the start.","triggerScenarios":"Calling Open() twice on the same singleUseReStream, e.g. reading an element's value in two different exec nodes or retrying a consumption path that already opened the stream.","commonSituations":"Custom exec transform code that caches a FullValue and re-reads it; misconfiguring a dedup/caching layer that opens streams twice; debugging the exec package with repeated Open calls.","solutions":["Open the ReStream exactly once and keep the returned Stream for all reads.","If the data must be consumed multiple times, wrap the source so each consumer gets a fresh singleUseReStream.","Refactor the pipeline so the decoded element is shared instead of the stream."],"exampleFix":null,"handlingStrategy":"try-catch","validationCode":"// Track whether the stream was already opened\nif stream.opened { return errors.New(\"singleUseReStream already consumed\") }","typeGuard":"func isOpen(n *exec.singleUseReStream) bool { return n != nil /* and not yet opened; inspect r via package-local code */ }","tryCatchPattern":"s, err := rStream.Open()\nif err != nil {\n    // stream already consumed: re-materialize or obtain a fresh ReStream\n    return fmt.Errorf(\"stream unusable: %w\", err)\n}","preventionTips":["Open each single-use ReStream exactly once, in one place","Pass decoded values, not streams, to multiple consumers","Never cache the ReStream for later re-reading"],"tags":["go","apache-beam","exec","restream"],"backgroundTag":"invalid-state-transition","analyzedSha":"12126d8942aaf848030c478b4c6a28c6af861c66","analyzedAt":"2026-09-13T01:50:10.254Z","contentChangedAt":"2026-09-13T01:50:10.254Z","schemaVersion":2},"datasetVersion":"2026-09-20T03:17:13.778Z"}