func (ps *PushSync) handler(ctx context.Context, p p2p.Peer, stream p2p.Stream) (err error) {
	now := time.Now()
	w, r := protobuf.NewWriterAndReader(stream)
	ctx, cancel := context.WithTimeout(ctx, defaultTTL)
	defer cancel()
	defer func() {
		if err != nil {
			ps.metrics.TotalHandlerTime.WithLabelValues("failure").Observe(time.Since(now).Seconds())
			ps.metrics.TotalHandlerErrors.Inc()
			_ = stream.Reset()
		} else {
			ps.metrics.TotalHandlerTime.WithLabelValues("success").Observe(time.Since(now).Seconds())
			_ = stream.FullClose()
		}
	}()
	var ch pb.Delivery
	if err = r.ReadMsgWithContext(ctx, &ch); err != nil {
		return fmt.Errorf("pushsync read delivery: %w", err)
	}
	ps.metrics.TotalReceived.Inc()
	chunk := swarm.NewChunk(swarm.NewAddress(ch.Address), ch.Data)
	chunkAddress := chunk.Address()
	span, _, ctx := ps.tracer.StartSpanFromContext(ctx, "pushsync-handler", ps.logger, opentracing.Tag{Key: "address", Value: chunkAddress.String()})
	defer span.Finish()
	stamp := new(postage.Stamp)
	err = stamp.UnmarshalBinary(ch.Stamp)
	if err != nil {
		return fmt.Errorf("pushsync stamp unmarshall: %w", err)
	}
	chunk.WithStamp(stamp)
	if cac.Valid(chunk) {
		if ps.unwrap != nil {
			go ps.unwrap(chunk)
		}
	} else if !soc.Valid(chunk) {
		return swarm.ErrInvalidChunk
	}
	price := ps.pricer.Price(chunkAddress)
	if p.FullNode {
		if closer, _ := p.Address.Closer(chunkAddress, ps.address); closer {
			ps.metrics.HandlerReplication.Inc()
			ctxd, canceld := context.WithTimeout(context.Background(), timeToWaitForPushsyncToNeighbor)
			defer canceld()
			span, _, ctxd := ps.tracer.StartSpanFromContext(ctxd, "pushsync-replication-storage", ps.logger, opentracing.Tag{Key: "address", Value: chunkAddress.String()})
			defer span.Finish()
			realClosestPeer, err := ps.topologyDriver.ClosestPeer(chunk.Address(), false)
			if err == nil {
				if !realClosestPeer.Equal(p.Address) {
					ps.metrics.TotalReplicationFromDistantPeer.Inc()
				} else {
					ps.metrics.TotalReplicationFromClosestPeer.Inc()
				}
			}
			chunk, err = ps.validStamp(chunk, ch.Stamp)
			if err != nil {
				ps.metrics.InvalidStampErrors.Inc()
				ps.metrics.HandlerReplicationErrors.Inc()
				return fmt.Errorf("pushsync replication valid stamp: %w", err)
			}
			_, err = ps.storer.Put(ctxd, storage.ModePutSync, chunk)
			if err != nil {
				ps.metrics.HandlerReplicationErrors.Inc()
				return fmt.Errorf("chunk store: %w", err)
			}
			debit, err := ps.accounting.PrepareDebit(p.Address, price)
			if err != nil {
				ps.metrics.HandlerReplicationErrors.Inc()
				return fmt.Errorf("prepare debit to peer %s before writeback: %w", p.Address.String(), err)
			}
			defer debit.Cleanup()
			signature, err := ps.signer.Sign(chunkAddress.Bytes())
			if err != nil {
				ps.metrics.HandlerReplicationErrors.Inc()
				return fmt.Errorf("receipt signature: %w", err)
			}
			receipt := pb.Receipt{Address: chunkAddress.Bytes(), Signature: signature, BlockHash: ps.blockHash}
			if err := w.WriteMsgWithContext(ctxd, &receipt); err != nil {
				ps.metrics.HandlerReplicationErrors.Inc()
				return fmt.Errorf("send receipt to peer %s: %w", p.Address.String(), err)
			}
			err = debit.Apply()
			if err != nil {
				ps.metrics.HandlerReplicationErrors.Inc()
			}
			return err
		}
	}
	storedChunk := false
	if ps.warmedUp() && ps.topologyDriver.IsWithinDepth(chunkAddress) {
		verifiedChunk, err := ps.validStamp(chunk, ch.Stamp)
		if err != nil {
			ps.metrics.InvalidStampErrors.Inc()
			ps.logger.Warningf("pushsync: forwarder, invalid stamp for chunk %s", chunkAddress.String())
		} else {
			chunk = verifiedChunk
			_, err = ps.storer.Put(ctx, storage.ModePutSync, chunk)
			if err != nil {
				ps.logger.Warningf("pushsync: within depth peer's attempt to store chunk failed: %v", err)
			} else {
				storedChunk = true
			}
		}
	}
	receipt, err := ps.pushToClosest(ctx, chunk, false, p.Address)
	if err != nil {
		if errors.Is(err, topology.ErrWantSelf) {
			ps.metrics.Storer.Inc()
			if !storedChunk {
				chunk, err = ps.validStamp(chunk, ch.Stamp)
				if err != nil {
					ps.metrics.InvalidStampErrors.Inc()
					return fmt.Errorf("pushsync storer valid stamp: %w", err)
				}
				_, err = ps.storer.Put(ctx, storage.ModePutSync, chunk)
				if err != nil {
					return fmt.Errorf("chunk store: %w", err)
				}
			}
			signature, err := ps.signer.Sign(ch.Address)
			if err != nil {
				return fmt.Errorf("receipt signature: %w", err)
			}
			debit, err := ps.accounting.PrepareDebit(p.Address, price)
			if err != nil {
				return fmt.Errorf("prepare debit to peer %s before writeback: %w", p.Address.String(), err)
			}
			defer debit.Cleanup()
			receipt := pb.Receipt{Address: chunkAddress.Bytes(), Signature: signature, BlockHash: ps.blockHash}
			if err := w.WriteMsgWithContext(ctx, &receipt); err != nil {
				return fmt.Errorf("send receipt to peer %s: %w", p.Address.String(), err)
			}
			return debit.Apply()
		}
		ps.metrics.Forwarder.Inc()
		return fmt.Errorf("handler: push to closest: %w", err)
	}
	ps.metrics.Forwarder.Inc()
	debit, err := ps.accounting.PrepareDebit(p.Address, price)
	if err != nil {
		return fmt.Errorf("prepare debit to peer %s before writeback: %w", p.Address.String(), err)
	}
	defer debit.Cleanup()
	if err := w.WriteMsgWithContext(ctx, receipt); err != nil {
		return fmt.Errorf("send receipt to peer %s: %w", p.Address.String(), err)
	}
	return debit.Apply()
}
handler handles chunk delivery from other node and forwards to its destination node.
If the current node is the destination, it stores in the local store and sends a receipt.
package pushsync
import (
	"context"
	"errors"
	"fmt"
	"sync"
	"time"
	"github.com/ethersphere/bee/pkg/accounting"
	"github.com/ethersphere/bee/pkg/cac"
	"github.com/ethersphere/bee/pkg/crypto"
	"github.com/ethersphere/bee/pkg/logging"
	"github.com/ethersphere/bee/pkg/p2p"
	"github.com/ethersphere/bee/pkg/p2p/protobuf"
	"github.com/ethersphere/bee/pkg/postage"
	"github.com/ethersphere/bee/pkg/pricer"
	"github.com/ethersphere/bee/pkg/pushsync/pb"
	"github.com/ethersphere/bee/pkg/soc"
	"github.com/ethersphere/bee/pkg/storage"
	"github.com/ethersphere/bee/pkg/swarm"
	"github.com/ethersphere/bee/pkg/tags"
	"github.com/ethersphere/bee/pkg/topology"
	"github.com/ethersphere/bee/pkg/tracing"
	opentracing "github.com/opentracing/opentracing-go"
)
const (
	protocolName    = "pushsync"
	protocolVersion = "1.0.0"
	streamName      = "pushsync"
)
const (
	maxPeers    = 3
	maxAttempts = 16
)
var (
	ErrOutOfDepthReplication = errors.New("replication outside of the neighborhood")
	ErrNoPush                = errors.New("could not push chunk")
	ErrOutOfDepthStoring     = errors.New("storing outside of the neighborhood")
	ErrWarmup                = errors.New("node warmup time not complete")
	defaultTTL                      = 20 * time.Second  
	sanctionWait                    = 5 * time.Minute
	timeToWaitForPushsyncToNeighbor = 5 * time.Second  
	nPeersToPushsync                = 3                
)
type PushSyncer interface {
	PushChunkToClosest(ctx context.Context, ch swarm.Chunk) (*Receipt, error)
}
type Receipt struct {
	Address   swarm.Address
	Signature []byte
	BlockHash []byte
}
type PushSync struct {
	address        swarm.Address
	blockHash      []byte
	streamer       p2p.StreamerDisconnecter
	storer         storage.Putter
	topologyDriver topology.Driver
	tagger         *tags.Tags
	unwrap         func(swarm.Chunk)
	logger         logging.Logger
	accounting     accounting.Interface
	pricer         pricer.Interface
	metrics        metrics
	tracer         *tracing.Tracer
	validStamp     postage.ValidStampFn
	signer         crypto.Signer
	isFullNode     bool
	warmupPeriod   time.Time
	skipList       *peerSkipList
}
func New(address swarm.Address, blockHash []byte, streamer p2p.StreamerDisconnecter, storer storage.Putter, topology topology.Driver, tagger *tags.Tags, isFullNode bool, unwrap func(swarm.Chunk), validStamp postage.ValidStampFn, logger logging.Logger, accounting accounting.Interface, pricer pricer.Interface, signer crypto.Signer, tracer *tracing.Tracer, warmupTime time.Duration) *PushSync {
	ps := &PushSync{
		address:        address,
		blockHash:      blockHash,
		streamer:       streamer,
		storer:         storer,
		topologyDriver: topology,
		tagger:         tagger,
		isFullNode:     isFullNode,
		unwrap:         unwrap,
		logger:         logger,
		accounting:     accounting,
		pricer:         pricer,
		metrics:        newMetrics(),
		tracer:         tracer,
		validStamp:     validStamp,
		signer:         signer,
		skipList:       newPeerSkipList(),
		warmupPeriod:   time.Now().Add(warmupTime),
	}
	return ps
}
func (s *PushSync) Protocol() p2p.ProtocolSpec {
	return p2p.ProtocolSpec{
		Name:    protocolName,
		Version: protocolVersion,
		StreamSpecs: []p2p.StreamSpec{
			{
				Name:    streamName,
				Handler: s.handler,
			},
		},
	}
}