{"record":{"id":"bea8d4d39b0282fe","repo":"dgraph-io/badger","slug":"write-performed-on-closed-stream-d","errorCode":null,"errorMessage":"write performed on closed stream: %d","messagePattern":"write performed on closed stream: (.+?)","errorType":"panic","errorClass":null,"httpStatus":null,"severity":"critical","filePath":"stream_writer.go","lineNumber":154,"sourceCode":"\t// closedStreams keeps track of all streams which are going to be marked as done. We are\n\t// keeping track of all streams so that we can close them at the end, after inserting all\n\t// the valid kvs.\n\tclosedStreams := make(map[uint32]struct{})\n\tstreamReqs := make(map[uint32]*request)\n\n\terr := buf.SliceIterate(func(s []byte) error {\n\t\tvar kv pb.KV\n\t\tif err := proto.Unmarshal(s, &kv); err != nil {\n\t\t\treturn err\n\t\t}\n\t\tif kv.StreamDone {\n\t\t\tclosedStreams[kv.StreamId] = struct{}{}\n\t\t\treturn nil\n\t\t}\n\n\t\t// Panic if some kv comes after stream has been marked as closed.\n\t\tif _, ok := closedStreams[kv.StreamId]; ok {\n\t\t\tpanic(fmt.Sprintf(\"write performed on closed stream: %d\", kv.StreamId))\n\t\t}\n\n\t\tsw.writeLock.Lock()\n\t\tif sw.maxVersion < kv.Version {\n\t\t\tsw.maxVersion = kv.Version\n\t\t}\n\t\tif sw.prevLevel == 0 {\n\t\t\t// If prevLevel is 0, that means that we have not written anything yet.\n\t\t\t// So, we can write to the maxLevel. newWriter writes to prevLevel - 1,\n\t\t\t// so we can set prevLevel to len(levels).\n\t\t\tsw.prevLevel = len(sw.db.lc.levels)\n\t\t}\n\t\tsw.writeLock.Unlock()\n\n\t\tvar meta, userMeta byte\n\t\tif len(kv.Meta) > 0 {\n\t\t\tmeta = kv.Meta[0]\n\t\t}","sourceCodeStart":136,"sourceCodeEnd":172,"githubUrl":"https://github.com/dgraph-io/badger/blob/2a001d466f6b71a917319a1db41f99860e16e269/stream_writer.go#L136-L172","documentation":"The StreamWriter callback panics if a key-value arrives for a stream that was already marked as done (closed). Sending KVs after SendDone for a stream ID indicates a producer bug and would corrupt the ordering assumptions of the writer, so it panics immediately.","triggerScenarios":"A Send/Orchestrate implementation calls sw.Send(kv) for a stream ID after having called sw.SendDone(id) for that same ID — e.g., sending duplicate iterations of the same stream, reusing a closed stream ID, or racing multiple goroutines over one stream ID.","commonSituations":"Custom stream.Send implementations that iterate sources concurrently and close one stream while another goroutine still emits to it; re-running Orchestrate with cached/old buffers containing KVs of closed streams.","solutions":["Ensure each stream ID emits all KVs before calling sw.SendDone(id) exactly once","Do not reuse a stream ID after SendDone; create a new stream ID for further data","Synchronize producers so a single goroutine (or a properly coordinated set) owns all sends for a given stream ID"],"exampleFix":"// before\nsw.SendDone(kv.StreamId) // closes stream\n// ...later...\nsw.Send(kv) // panics: write performed on closed stream\n// after\nsw.Send(kv) // send all KVs first\nsw.SendDone(kv.StreamId) // then close, and never send to this ID again","handlingStrategy":"validation","validationCode":"// In your stream.Send implementation, never call sw.Send after SendDone for a stream ID.\nfunc (p *producer) send(sw *badger.StreamWriter, kv *badger.KV) error {\n    if p.closed[kv.StreamId] { return fmt.Errorf(\"stream %d already done\", kv.StreamId) }\n    return sw.Send(kv)\n}","typeGuard":"func streamClosed(closed map[uint32]struct{}, id uint32) bool {\n    _, ok := closed[id]\n    return ok\n}","tryCatchPattern":null,"preventionTips":["Call SendDone(id) exactly once per stream, strictly after all Send calls for that ID","Never reuse a stream ID after SendDone; allocate a fresh ID for subsequent data","Serialize sends per stream ID (one producer goroutine or a mutex) to avoid post-close races"],"tags":["badger","panic","streamwriter","concurrency"],"backgroundTag":"write-after-close","analyzedSha":"2a001d466f6b71a917319a1db41f99860e16e269","analyzedAt":"2026-09-05T13:00:02.264Z","contentChangedAt":"2026-09-05T13:00:02.264Z","schemaVersion":2},"datasetVersion":"2026-09-12T17:17:11.597Z"}