func (t *StoreHost) SealExtent(ctx thrift.Context, req *store.SealExtentRequest) (err error) {
	t.m3Client.IncCounter(metrics.SealExtentScope, metrics.StorageRequests)
	if atomic.LoadInt32(&t.started) == 0 {
		t.m3Client.IncCounter(metrics.SealExtentScope, metrics.StorageFailures)
		return newInternalServiceError("StoreHost not started")
	}
	var sealSeqNum, unspecifiedSealSeqNum = seqNumUnspecifiedSeal, true
	if req.IsSetSequenceNumber() {
		sealSeqNum = req.GetSequenceNumber()
		unspecifiedSealSeqNum = false
	}
	log := t.logger.WithField(common.TagExt, common.FmtExt(req.GetExtentUUID()))
	log.WithField("args", fmt.Sprintf("sealSeqNum=%d", sealSeqNum)).
		Info("SealExtent()")
	if !t.sealExtentThrottler.Allow() {
		log.Error("SealExtent failed: request throttled")
		t.m3Client.IncCounter(metrics.SealExtentScope, metrics.StorageFailures)
		return newInternalServiceError("request throttled")
	}
	sw := t.m3Client.StartTimer(metrics.SealExtentScope, metrics.StorageLatencyTimer)
	defer sw.Stop()
	extentID := uuid.Parse(req.GetExtentUUID())
	if extentID == nil {
		log.Error("SealExtent failed: error parsing extentuuid")
		t.m3Client.IncCounter(metrics.SealExtentScope, metrics.StorageFailures)
		return newBadRequestError(fmt.Sprintf("error parsing extentuuid (%s)", req.GetExtentUUID()))
	}
	x, err := t.xMgr.OpenExtent(extentID, AppendOnly, OpenIntentSealExtent)
	if err != nil {
		log.WithField("context", fmt.Sprintf("err=%v", err)).
			Error("SealExtent failed: OpenExtent error")
		t.m3Client.IncCounter(metrics.SealExtentScope, metrics.StorageFailures)
		t.m3Client.IncCounter(metrics.SealExtentScope, metrics.StorageStoreFailures)
		return newInternalServiceError(fmt.Sprintf("%v OpenExtent error: %v", extentID, err))
	}
	defer x.Close()  
	defer x.storeSync()
	if oldSealSeqNum := x.getSealSeqNum(); oldSealSeqNum != seqNumNotSealed {  
		switch {
		case sealSeqNum == seqNumNotSealed:  
			log.WithField("context", fmt.Sprintf("oldSealSeqNum=%d", oldSealSeqNum)).
				Error("SealExtent: extent already sealed")
		case sealSeqNum > oldSealSeqNum:  
			log.WithField("context", fmt.Sprintf("oldSealSeqNum=%d sealSeqNum=%d", oldSealSeqNum, sealSeqNum)).
				Error("SealExtent: extent sealed at seqNumber less than requested")
		case sealSeqNum == oldSealSeqNum:  
			log.WithField("context", fmt.Sprintf("sealSeqNum=%d", sealSeqNum)).
				Info("SealExtent: extent already sealed at requested seqnum")
			return nil
		default:  
		}
		x.extentLock()
		defer x.extentUnlock()
		oldSealSeqNum = x.getSealSeqNum()
		if sealSeqNum >= oldSealSeqNum {
			if sealSeqNum == oldSealSeqNum {  
				log.WithField("context", fmt.Sprintf("sealSeqNum=%d", sealSeqNum)).
					Info("SealExtent: extent already sealed at requested seqnum")
				return nil  
			}
			if sealSeqNum > oldSealSeqNum {  
				log.WithField("context", fmt.Sprintf("oldSealSeqNum=%d sealSeqNum=%d", oldSealSeqNum, sealSeqNum)).
					Error("SealExtent: [2] extent sealed at seqNumber less than requested")
			}
		}
	} else {  
		x.extentLock()
		defer x.extentUnlock()
		lastSeqNum := x.getLastSeqNum()
		switch {
		case lastSeqNum == SeqnumInvalid && unspecifiedSealSeqNum:  
		case lastSeqNum == SeqnumInvalid && !unspecifiedSealSeqNum:  
			log.Error("SealExtent: last-seqnum unknown (timer-queue mode?)")
		case lastSeqNum != SeqnumInvalid && unspecifiedSealSeqNum:  
			sealSeqNum = lastSeqNum  
		case lastSeqNum != SeqnumInvalid && !unspecifiedSealSeqNum:  
			if sealSeqNum > lastSeqNum {
				log.WithField("context", fmt.Sprintf("lastSeqNum=%d sealSeqNum=%d", lastSeqNum, sealSeqNum)).Error("SealExtent: last-seqnum less than seal-seqnum")
			}
			if sealSeqNum < lastSeqNum {
				log.WithField("context", fmt.Sprintf("lastSeqNum=%d sealSeqNum=%d", lastSeqNum, sealSeqNum)).Error("sealSeqNum < lastSeqNum: potentially un-acked messages")
			}
		}
	}
	sealExtentKey := x.constructSealExtentKey(sealSeqNum)
	_, err = x.storePut(sealExtentKey, []byte{})
	if err != nil {
		log.WithField("context", fmt.Sprintf("err=%v", err)).
			Error("SealExtent: error writing seal-key")
		t.m3Client.IncCounter(metrics.SealExtentScope, metrics.StorageFailures)
		return newInternalServiceError(fmt.Sprintf("%v failed writing seal-key: %v", extentID, err))
	}
	x.setSealSeqNum(sealSeqNum)  
	if unspecifiedSealSeqNum {
		log.WithField("context", fmt.Sprintf("sealSeqNum=%d", sealSeqNum)).
			Info("SealExtent with unspecified seqnum sealed")
	}
	log.WithField("context", fmt.Sprintf("sealSeqNum=%d", sealSeqNum)).
		Info("SealExtent done")
	return nil
}
SealExtent is the implementation of the thrift handler for the store host
package storehost
import (
	"fmt"
	"net"
	"net/http"
	"os"
	"strconv"
	"sync"
	"sync/atomic"
	"time"
	ccommon "github.com/uber/cherami-client-go/common"
	"github.com/uber/cherami-server/common"
	mm "github.com/uber/cherami-server/common/metadata"
	"github.com/uber/cherami-server/common/metrics"
	"github.com/uber/cherami-server/common/throttler"
	"github.com/uber/cherami-server/storage"
	"github.com/uber/cherami-thrift/.generated/go/cherami"
	"github.com/uber/cherami-thrift/.generated/go/metadata"
	"github.com/uber/cherami-thrift/.generated/go/store"
	"github.com/uber/cherami-server/services/storehost/load"
	"github.com/uber/cherami-server/storage/manyrocks"
	storeStream "github.com/uber/cherami-server/stream"
	"github.com/uber/cherami-thrift/.generated/go/controller"
	"github.com/pborman/uuid"
	"github.com/uber-common/bark"
	"github.com/uber/tchannel-go/thrift"
)
const (
	sealCheckInterval time.Duration = 500 * time.Millisecond
)
const (
	storehostLoadReportingInterval = 2 * time.Second
)
const (
	writeMsgChanBuf int = 4096  
	writeAckChanBuf int = 1024  
	readMsgChanBuf  int = 1024  
)
const (
	sealExtentThrottleRequests               = 25
	sealExtentThrottlePeriod   time.Duration = 250 * time.Millisecond
)
type conn interface {
	Start()
	Stop()
	Done() <-chan struct{}
}
const (
	defaultStore   Store  = ManyRocks  
	defaultBaseDir string = "./CHERAMI_STORE"
)
type Mode int
const (
	_ Mode = iota  
	AppendOnly
	TimerQueue
	Log
)
func (t Mode) String() string {
	switch t {
	case AppendOnly:
		return "AppendOnly"
	case TimerQueue:
		return "TimerQueue"
	case Log:
		return "Log"
	default:
		return "UNKNOWN"
	}
}
type Store int
const (
	_ Store = iota  
	Rockstor
	Chunky
	ManyRocks
	RockCFstor
)
func (t Store) String() string {
	switch t {
	case Rockstor:
		return "Rockstor"
	case Chunky:
		return "Chunky"
	case ManyRocks:
		return "ManyRocks"
	case RockCFstor:
		return "RockCFstor"
	default:
		return "UNKNOWN"
	}
}
type (
	Options struct {
		Store   Store
		BaseDir string
	}
	StoreHost struct {
		common.SCommon
		opts              *Options
		mClient           metadata.TChanMetadataService
		hostIDHeartbeater common.HostIDHeartbeater
		loadReporter      common.LoadReporterDaemon
		shutdownWG sync.WaitGroup
		xMgr          *ExtentManager       
		replMgr       *ReplicationManager  
		shutdownC     chan struct{}
		disableWriteC chan struct{}
		numInConn, numOutConn int64  
		sealExtentThrottler *throttler.Throttler
		logger bark.Logger
		m3Client metrics.Client
		replicationJobRunner ReplicationJobRunner
		extStatsReporter *ExtStatsReporter
		storageMonitor StorageMonitor
		hostMetrics *load.HostMetrics
		lastLoadReportedTime int64
		nodeStatus atomic.Value
		started int32
	}
)
var _ store.TChanBStore = &StoreHost{}
var _ common.WSService = &StoreHost{}
func NewStoreHost(serviceName string, sCommon common.SCommon, mClient metadata.TChanMetadataService, opts *Options) (*StoreHost, []thrift.TChanServer) {
	if opts == nil {
		opts = &Options{}
	}
	logger := sCommon.GetConfig().GetLogger().WithFields(bark.Fields{
		common.TagStor:    common.FmtStor(sCommon.GetHostUUID()),
		common.TagDplName: common.FmtDplName(sCommon.GetConfig().GetDeploymentName()),
	})
	m3Client := metrics.NewClient(sCommon.GetMetricsReporter(), metrics.Storage)
	t := &StoreHost{
		SCommon:       sCommon,
		opts:          opts,
		hostMetrics:   load.NewHostMetrics(),
		shutdownC:     make(chan struct{}),
		disableWriteC: make(chan struct{}),
		logger:        logger,
		m3Client:      m3Client,
		mClient:       mm.NewMetadataMetricsMgr(mClient, m3Client, logger),
	}
	return t, []thrift.TChanServer{store.NewTChanBStoreServer(t)}
}
func (t *StoreHost) Start(thriftService []thrift.TChanServer) {
	t.SCommon.Start(thriftService)
	hostID := t.GetHostUUID()
	t.hostIDHeartbeater = common.NewHostIDHeartbeater(t.mClient, hostID, t.GetHostPort(), t.GetHostName(), t.logger)
	t.hostIDHeartbeater.Start()
	var storeMgr storage.StoreManager
	var baseDir string
	switch t.opts.Store {
	default:
		fallthrough
	case ManyRocks:
		var err error
		baseDir = fmt.Sprintf("%s/%s", t.opts.BaseDir, hostID)
		storeMgr, err = manyrocks.New(&manyrocks.Opts{BaseDir: baseDir}, t.logger)
		if err != nil {
			t.logger.WithField("options", fmt.Sprintf("Store=%v BaseDir=%v", t.opts.Store, t.opts.BaseDir)).
				Fatal("Error initializing ManyRocks")
			return
		}
	}
	t.sealExtentThrottler = throttler.New(sealExtentThrottleRequests, sealExtentThrottlePeriod)
	t.xMgr = NewExtentManager(storeMgr, t.m3Client, t.hostMetrics, t.logger)
	t.storageMonitor = NewStorageMonitor(t, t.m3Client, t.hostMetrics, t.logger, baseDir)
	t.storageMonitor.Start()
	t.replMgr = NewReplicationManager(t.xMgr, t.m3Client, t.mClient, t.logger, hostID, t.GetWSConnector())
	t.replicationJobRunner = NewReplicationJobRunner(t.mClient, t, t.logger, t.m3Client)
	go t.replicationJobRunner.Start()
	t.SetNodeStatus(controller.NodeStatus_UP)
	loadReporterDaemonfactory := t.GetLoadReporterDaemonFactory()
	t.xMgr.loadReporterFactory = loadReporterDaemonfactory
	t.loadReporter = loadReporterDaemonfactory.CreateReporter(storehostLoadReportingInterval, t, t.logger)
	t.lastLoadReportedTime = time.Now().UnixNano()
	t.loadReporter.Start()
	t.extStatsReporter = NewExtStatsReporter(hostID, t.xMgr, t.mClient, t.logger)
	t.extStatsReporter.Start()
	atomic.StoreInt32(&t.started, 1)  
	t.logger.WithField("options", fmt.Sprintf("Store=%v BaseDir=%v", t.opts.Store, t.opts.BaseDir)).
		Info("StoreHost: started")
}
func (t *StoreHost) Stop() {
	atomic.StoreInt32(&t.started, 0)  
	t.loadReporter.Stop()
	t.hostIDHeartbeater.Stop()
	t.storageMonitor.Stop()
	t.replicationJobRunner.Stop()
	t.extStatsReporter.Stop()
	t.SCommon.Stop()
	t.logger.Info("StoreHost: stopped")
}
func getInConnArgs(ctx thrift.Context) (args *inConnArgs, err error) {
	req, err := common.GetOpenAppendStreamRequestStruct(ctx.Headers())
	if err != nil {
		return nil, err
	}
	args = &inConnArgs{}  
	if args.extentID = uuid.Parse(req.GetExtentUUID()); args.extentID == nil {
		return nil, newBadRequestError(fmt.Sprintf("BadRequestError: error parsing uuid (%s)", req.GetExtentUUID()))
	}
	if args.destID = uuid.Parse(req.GetDestinationUUID()); args.destID == nil {
		return nil, newBadRequestError(fmt.Sprintf("BadRequestError: error parsing uuid (%s)", req.GetDestinationUUID()))
	}
	args.destType = req.GetDestinationType()
	args.mode = getModeForDestinationType(args.destType)
	if args.mode == 0 {
		return nil, newBadRequestError(fmt.Sprintf("BadRequestError: unknown type (%d)", args.mode))
	}
	return args, nil
}
func (t *StoreHost) OpenAppendStreamHandler(w http.ResponseWriter, r *http.Request) {
	req, err := common.GetOpenAppendStreamRequestHTTP(r.Header)
	if err != nil {
		t.logger.WithField(`error`, err).Error("unable to parse all needed headers")
		t.m3Client.IncCounter(metrics.OpenAppendStreamScope, metrics.StorageFailures)
		http.Error(w, err.Error(), http.StatusBadRequest)
		return
	}
	wsStream, err := t.GetWSConnector().AcceptAppendStream(w, r)
	if err != nil {
		t.logger.WithField(`error`, err).Error("unable to upgrade websocket connection")
		t.m3Client.IncCounter(metrics.OpenAppendStreamScope, metrics.StorageFailures)
		return
	}
	ctx, cancel := thrift.NewContext(common.MaxDuration)
	defer cancel()
	ctx = thrift.WithHeaders(ctx, common.GetOpenAppendStreamRequestHeaders(req))
	if err = t.OpenAppendStream(ctx, wsStream); err != nil {
		t.logger.WithField(`error`, err).Error("unable to open append stream")
		return
	}
}
func (t *StoreHost) OpenAppendStream(ctx thrift.Context, call storeStream.BStoreOpenAppendStreamInCall) error {
	t.m3Client.IncCounter(metrics.OpenAppendStreamScope, metrics.StorageRequests)
	if atomic.LoadInt32(&t.started) == 0 {
		call.Done()
		t.m3Client.IncCounter(metrics.OpenAppendStreamScope, metrics.StorageFailures)
		return newInternalServiceError("StoreHost not started")
	}
	if t.storageMonitor != nil && t.storageMonitor.GetStorageMode() == SMReadOnly {
		call.Done()
		t.m3Client.IncCounter(metrics.OpenAppendStreamScope, metrics.StorageFailures)
		return newInternalServiceError("StoreHost in read-only mode")
	}
	args, err := getInConnArgs(ctx)
	if err != nil {
		call.Done()
		t.m3Client.IncCounter(metrics.OpenAppendStreamScope, metrics.StorageFailures)
		return err
	}
	log := t.logger.WithFields(bark.Fields{
		common.TagExt: common.FmtExt(args.extentID.String()),
		common.TagDst: common.FmtDst(args.destID.String()),
	})
	log.WithField("args", fmt.Sprintf("destType=%v mode=%v", args.destType, args.mode)).
		Info("OpenAppendStream: starting inConn")
	in := newInConn(args, call, t.xMgr, t.m3Client, log)
	t.shutdownWG.Add(1)
	defer t.shutdownWG.Done()
	in.Start()  
	numInConn := atomic.AddInt64(&t.numInConn, 1)
	t.m3Client.UpdateGauge(metrics.OpenAppendStreamScope, metrics.StorageWriteStreams, numInConn)
	log.WithField(`numInConn`, numInConn).Info("OpenAppendStream: write stream opened")
	select {
	case err = <-in.Done():
	case <-t.shutdownC:
		err = in.Stop()  
	case <-t.disableWriteC:
		log.Info("Stop write due to available disk space is extremely low")
		err = in.Stop()
	}
	numInConn = atomic.AddInt64(&t.numInConn, -1)
	t.m3Client.UpdateGauge(metrics.OpenAppendStreamScope, metrics.StorageWriteStreams, numInConn)
	log.WithField(`numInConn`, numInConn).Info("OpenAppendStream: write stream closed")
	log.Info("OpenAppendStream done")
	return err  
}
func getOutConnArgs(ctx thrift.Context) (args *outConnArgs, err error) {
	req, err := common.GetOpenReadStreamRequestStruct(ctx.Headers())
	if err != nil {
		return nil, err
	}
	args = &outConnArgs{}  
	if args.extentID = uuid.Parse(req.GetExtentUUID()); args.extentID == nil {
		return nil, newBadRequestError(fmt.Sprintf("error parsing uuid (%s)", req.GetExtentUUID()))
	}
	if args.destID = uuid.Parse(req.GetDestinationUUID()); args.destID == nil {
		return nil, newBadRequestError(fmt.Sprintf("BadRequestError: error parsing uuid (%s)", req.GetDestinationUUID()))
	}
	args.destType = req.GetDestinationType()
	args.mode = getModeForDestinationType(args.destType)
	if args.mode == 0 {
		return nil, newBadRequestError(fmt.Sprintf("BadRequestError: unknown type (%d)", args.mode))
	}
	if args.consGroupID = uuid.Parse(req.GetConsumerGroupUUID()); args.consGroupID == nil {
		return nil, newBadRequestError(fmt.Sprintf("BadRequestError: error parsing uuid (%s)", req.GetConsumerGroupUUID()))
	}
	args.address = req.GetAddress()
	args.inclusive = req.GetInclusive()
	return args, nil
}
func (t *StoreHost) OpenReadStreamHandler(w http.ResponseWriter, r *http.Request) {
	req, err := common.GetOpenReadStreamRequestHTTP(r.Header)
	if err != nil {
		t.logger.WithField(`error`, err).Error("unable to parse all needed headers")
		t.m3Client.IncCounter(metrics.OpenReadStreamScope, metrics.StorageFailures)
		http.Error(w, err.Error(), http.StatusBadRequest)
		return
	}
	wsStream, err := t.GetWSConnector().AcceptReadStream(w, r)
	if err != nil {
		t.logger.WithField(`error`, err).Error("unable to upgrade websocket connection")
		t.m3Client.IncCounter(metrics.OpenReadStreamScope, metrics.StorageFailures)
		return
	}
	ctx, cancel := thrift.NewContext(common.MaxDuration)
	defer cancel()
	ctx = thrift.WithHeaders(ctx, common.GetOpenReadStreamRequestHeaders(req))
	if err = t.OpenReadStream(ctx, wsStream); err != nil {
		t.logger.WithField(`error`, err).Error("unable to open read stream")
		return
	}
}
func (t *StoreHost) OpenReadStream(ctx thrift.Context, call storeStream.BStoreOpenReadStreamInCall) error {
	t.m3Client.IncCounter(metrics.OpenReadStreamScope, metrics.StorageRequests)
	if atomic.LoadInt32(&t.started) == 0 {
		call.Done()
		t.m3Client.IncCounter(metrics.OpenReadStreamScope, metrics.StorageFailures)
		return newInternalServiceError("StoreHost not started")
	}
	args, e := getOutConnArgs(ctx)
	if e != nil {
		err := newReadMessageError(e.Error())
		t.logger.WithField(common.TagErr, err).
			Error("OpenReadStream: getReadStreamArgs error")
		if e = call.Write(err); e != nil {
			t.logger.WithField(common.TagErr, err).
				Error("OpenReadStream: stream.Write error")
		}
		call.Done()
		t.m3Client.IncCounter(metrics.OpenReadStreamScope, metrics.StorageFailures)
		return nil  
	}
	log := t.logger.WithFields(bark.Fields{
		common.TagExt:  common.FmtExt(args.extentID.String()),
		common.TagDst:  common.FmtDst(args.destID.String()),
		common.TagCnsm: common.FmtCnsm(args.consGroupID.String()),
	})
	out := newOutConn(args, call, t.xMgr, t.m3Client, log)
	t.shutdownWG.Add(1)
	defer t.shutdownWG.Done()
	out.Start()  
	numOutConn := atomic.AddInt64(&t.numOutConn, 1)
	t.m3Client.UpdateGauge(metrics.OpenReadStreamScope, metrics.StorageReadStreams, numOutConn)
	log.WithFields(bark.Fields{
		`destType`:   args.destType,
		`mode`:       args.mode,
		`address`:    args.address,
		`inclusive`:  args.inclusive,
		`numOutConn`: numOutConn,
	}).Info("OpenReadStream: outConn started")
	select {
	case <-out.Done():
	case <-t.shutdownC:
		out.Stop()  
	}
	numOutConn = atomic.AddInt64(&t.numOutConn, -1)
	t.m3Client.UpdateGauge(metrics.OpenReadStreamScope, metrics.StorageReadStreams, numOutConn)
	log.WithField(`numOutConn`, numOutConn).Info("OpenReadStream: outConn done")
	return nil  
}
func (t *StoreHost) GetAddressFromTimestamp(ctx thrift.Context, req *store.GetAddressFromTimestampRequest) (res *store.GetAddressFromTimestampResult_, err error) {
	t.m3Client.IncCounter(metrics.GetAddressFromTimestampScope, metrics.StorageRequests)
	if atomic.LoadInt32(&t.started) == 0 {
		t.m3Client.IncCounter(metrics.GetAddressFromTimestampScope, metrics.StorageFailures)
		return nil, newInternalServiceError("StoreHost not started")
	}
	sw := t.m3Client.StartTimer(metrics.GetAddressFromTimestampScope, metrics.StorageLatencyTimer)
	defer sw.Stop()
	log := t.logger.WithField(common.TagExt, common.FmtExt(req.GetExtentUUID()))
	log.WithField("args", fmt.Sprintf("timestamp=%x", req.GetTimestamp())).
		Info("GetAddressFromTimestamp")
	extentID := uuid.Parse(req.GetExtentUUID())
	timestamp := req.GetTimestamp()
	if extentID == nil {
		log.Error("GetAddressFromTimestamp failed: error parsing uuid")
		t.m3Client.IncCounter(metrics.GetAddressFromTimestampScope, metrics.StorageFailures)
		return nil, newBadRequestError(fmt.Sprintf("error parsing uuid (%s)", req.GetExtentUUID()))
	}
	x, err := t.xMgr.OpenExtent(extentID, AppendOnly, OpenIntentGetAddressFromTimestamp)
	if err != nil {
		log.WithField("context", fmt.Sprintf("err=%v", err)).
			Error("GetAddressFromTimestamp failed: OpenExtent error")
		t.m3Client.IncCounter(metrics.GetAddressFromTimestampScope, metrics.StorageStoreFailures)
		t.m3Client.IncCounter(metrics.GetAddressFromTimestampScope, metrics.StorageFailures)
		return nil, newExtentNotFoundError(extentID, fmt.Sprintf("OpenExtent error: %v", err))
	}
	defer x.Close()  
	timeKey := x.constructKey(timestamp, 0)
	addr, addrKey, err := x.storeSeekFloor(timeKey)
	if err != nil {
		log.WithField("context", fmt.Sprintf("timeKey=%x err=%v", timeKey, err)).
			Error("GetAddressFromTimestamp failed SeekFloor()")
		t.m3Client.IncCounter(metrics.GetAddressFromTimestampScope, metrics.StorageStoreFailures)
		t.m3Client.IncCounter(metrics.GetAddressFromTimestampScope, metrics.StorageFailures)
		return nil, newInternalServiceError(fmt.Sprintf("%v store.SeekFloor(%x) failed: %v", extentID, timeKey, err))
	}
	_, nextKey, err := x.storeNext(addr)
	if err != nil {
		log.WithField("context", fmt.Sprintf("addr=%x err=%v", addr, err)).
			Error("GetAddressFromTimestamp failed Next()")
		t.m3Client.IncCounter(metrics.GetAddressFromTimestampScope, metrics.StorageStoreFailures)
		t.m3Client.IncCounter(metrics.GetAddressFromTimestampScope, metrics.StorageFailures)
		return nil, newInternalServiceError(fmt.Sprintf("%v store.Next(%x) failed: %v", extentID, addr, err))
	}
	res = store.NewGetAddressFromTimestampResult_()
	if addr == storage.MinAddr {
		res.Address = common.Int64Ptr(store.ADDR_BEGIN)
		res.SequenceNumber = common.Int64Ptr(0)
	} else {
		_, seqNum := x.deconstructKey(addrKey)
		res.Address = common.Int64Ptr(int64(addr))
		res.SequenceNumber = common.Int64Ptr(seqNum)
	}
	if nextKey != storage.InvalidKey && x.isSealExtentKey(nextKey) {
		res.Sealed = common.BoolPtr(true)
	} else {
		res.Sealed = common.BoolPtr(false)
	}
	log.WithField("return", fmt.Sprintf("timestamp=%v, address=%x, seq=%d, sealed=%v", req.GetTimestamp(), res.GetAddress(), res.GetSequenceNumber(), res.GetSealed())).
		Info("GetAddressFromTimestamp done")
	return res, nil
}