What causes "write performed on closed stream: %d" in dgraph-io/badger?

Ensure each stream ID emits all KVs before calling sw.SendDone(id) exactly once

write performed on closed stream: %d

Error message

write performed on closed stream: %d

What it means

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.

Source

Thrown at stream_writer.go:154

	// closedStreams keeps track of all streams which are going to be marked as done. We are
	// keeping track of all streams so that we can close them at the end, after inserting all
	// the valid kvs.
	closedStreams := make(map[uint32]struct{})
	streamReqs := make(map[uint32]*request)

	err := buf.SliceIterate(func(s []byte) error {
		var kv pb.KV
		if err := proto.Unmarshal(s, &kv); err != nil {
			return err
		}
		if kv.StreamDone {
			closedStreams[kv.StreamId] = struct{}{}
			return nil
		}

		// Panic if some kv comes after stream has been marked as closed.
		if _, ok := closedStreams[kv.StreamId]; ok {
			panic(fmt.Sprintf("write performed on closed stream: %d", kv.StreamId))
		}

		sw.writeLock.Lock()
		if sw.maxVersion < kv.Version {
			sw.maxVersion = kv.Version
		}
		if sw.prevLevel == 0 {
			// If prevLevel is 0, that means that we have not written anything yet.
			// So, we can write to the maxLevel. newWriter writes to prevLevel - 1,
			// so we can set prevLevel to len(levels).
			sw.prevLevel = len(sw.db.lc.levels)
		}
		sw.writeLock.Unlock()

		var meta, userMeta byte
		if len(kv.Meta) > 0 {
			meta = kv.Meta[0]
		}

View on GitHub (pinned to 2a001d466f)

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

Example fix

// before
sw.SendDone(kv.StreamId) // closes stream
// ...later...
sw.Send(kv) // panics: write performed on closed stream
// after
sw.Send(kv) // send all KVs first
sw.SendDone(kv.StreamId) // then close, and never send to this ID again

Defensive patterns

Strategy: validation

Validate before calling

// In your stream.Send implementation, never call sw.Send after SendDone for a stream ID.
func (p *producer) send(sw *badger.StreamWriter, kv *badger.KV) error {
    if p.closed[kv.StreamId] { return fmt.Errorf("stream %d already done", kv.StreamId) }
    return sw.Send(kv)
}

Type guard

func streamClosed(closed map[uint32]struct{}, id uint32) bool {
    _, ok := closed[id]
    return ok
}

Prevention

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

When it happens

Trigger: 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.

Common situations: 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.

Related errors

AI-assisted analysis of dgraph-io/badger@2a001d466f (2026-09-05). Data as JSON: /api/errors/bea8d4d39b0282fe. Report an issue: GitHub.