dgraph-io/dgraph · error

downstream( ) closed before Finish=true

Error message

downstream(%d) closed before Finish=true

What it means

After half-closing the downstream send, the relay drains the leader's responses waiting for the frame with Finish=true. If out.Recv() returns io.EOF instead, the leader closed its side without signalling completion, so the import result is unknown and this error is returned.

Solutions

1. Check the leader's logs for the error that ended its handler before Finish
2. Re-run the import; the snapshot state may be incomplete
3. Ensure the leader has adequate disk/memory for finalizing an external snapshot
4. Remove idle-timeout intermediaries between the two alphas

// ensure leader resources before import so finalization can complete
precheckDiskAndMemory(leaderHost)

func finishedCorrectly(resp *api.StreamExtSnapshotResponse) bool {
    return resp != nil && resp.Finish // only Finish=true means success
}

if err != nil && strings.Contains(err.Error(), "closed before Finish=true") {
    // outcome unknown: check leader logs, then re-run the import
}

When it happens

Trigger: The leader's StreamExtSnapshot server handler returned or closed the stream before sending Finish=true: it errored on its own snapshot pipeline, was killed, or its context was cancelled.

Common situations: Leader hit an internal error (disk full, memory) mid-finalization; leader restarted at the very end of the import; a proxy closed the connection after inactivity during the leader's final flush.

Related errors

• unexpected empty request
• ack data downstream( )
• failed to establish stream with leader
• failed to run in group streaming
• failed to run publisher

AI-assisted analysis of dgraph-io/dgraph@759e242be6 (2026-09-01). Data as JSON: /api/errors/f22121962719395e. Report an issue: GitHub.

Appendix: source

Thrown at worker/import.go:427

		if req.Pkt == nil {
			return fmt.Errorf("unexpected empty request")
		}

		if req.Pkt.Done {
			// Forward Done, half-close downstream send.
			if err := out.Send(&api.StreamExtSnapshotRequest{Pkt: req.Pkt}); err != nil && !errors.Is(err, io.EOF) {
				return fmt.Errorf("send done downstream(%d): %w", groupId, err)
			}
			_ = out.CloseSend()

			// Drain downstream and relay upstream until Finish=true.
			for {
				if err := ctx.Err(); err != nil {
					return err
				}
				resp, err := out.Recv()
				if errors.Is(err, io.EOF) {
					return fmt.Errorf("downstream(%d) closed before Finish=true", groupId)
				}
				if err != nil {
					return fmt.Errorf("recv final downstream(%d): %w", groupId, err)
				}
				if err := in.Send(resp); err != nil {
					return fmt.Errorf("relay final upstream: %w", err)
				}
				if resp.Finish {
					glog.Infof("[import] [forward %d -> %d] finish", currentGroup, groupId)
					return nil
				}
			}
		}

		// Normal data chunk: send -> wait ack -> send upstream ack.
		if err := out.Send(&api.StreamExtSnapshotRequest{Pkt: req.Pkt}); err != nil {
			return fmt.Errorf("send data downstream(%d): %w", groupId, err)
		}

View on GitHub (pinned to 759e242be6)