head monitor

.gitignore

@@ -20,3 +20,5 @@
 # Go workspace file
 go.work
 go.work.sum
+
+/octez-ecad-sc

go.mod

@@ -0,0 +1,14 @@
+module github.com/ecadlabs/octez-ecad-sc
+
+go 1.22.2
+
+require github.com/ecadlabs/gotez/v2 v2.0.6
+
+require (
+	github.com/ecadlabs/pretty v0.0.0-20230412124801-f948fc689a04 // indirect
+	github.com/kr/text v0.2.0 // indirect
+	github.com/rogpeppe/go-internal v1.9.0 // indirect
+	github.com/sirupsen/logrus v1.9.3 // indirect
+	golang.org/x/crypto v0.7.0 // indirect
+	golang.org/x/sys v0.21.0 // indirect
+)

go.sum

@@ -0,0 +1,32 @@
+github.com/creack/pty v1.1.9/go.mod h1:oKZEueFk5CKHvIhNR5MUki03XCEU+Q6VDXinZuGJ33E=
+github.com/davecgh/go-spew v1.1.0/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
+github.com/davecgh/go-spew v1.1.1 h1:vj9j/u1bqnvCEfJOwUhtlOARqs3+rkHYY13jYWTU97c=
+github.com/davecgh/go-spew v1.1.1/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
+github.com/ecadlabs/gotez/v2 v2.0.6 h1:P7eQ2G+SO1tTV4NHnkdNlrOHxKDo1iF9m34HTLfS3b8=
+github.com/ecadlabs/gotez/v2 v2.0.6/go.mod h1:QypK0m1eDPmB9R7Uvgmsfm+JS7Z5Y6dIbIq1tMVYayU=
+github.com/ecadlabs/pretty v0.0.0-20230412124801-f948fc689a04 h1:7WdblGykGxtGGtchW4kzTaJJO8Fm+JKhLzhttOOWr9k=
+github.com/ecadlabs/pretty v0.0.0-20230412124801-f948fc689a04/go.mod h1:VApUlocsLMpp4hUXHxTTIlosebnwo0BM6e1hy78qTPM=
+github.com/kr/text v0.2.0 h1:5Nx0Ya0ZqY2ygV366QzturHI13Jq95ApcVaJBhpS+AY=
+github.com/kr/text v0.2.0/go.mod h1:eLer722TekiGuMkidMxC/pM04lWEeraHUUmBw8l2grE=
+github.com/pkg/diff v0.0.0-20210226163009-20ebb0f2a09e/go.mod h1:pJLUxLENpZxwdsKMEsNbx1VGcRFpLqf3715MtcvvzbA=
+github.com/pmezard/go-difflib v1.0.0 h1:4DBwDE0NGyQoBHbLQYPwSUPoCMWR5BEzIk/f1lZbAQM=
+github.com/pmezard/go-difflib v1.0.0/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4=
+github.com/rogpeppe/go-internal v1.9.0 h1:73kH8U+JUqXU8lRuOHeVHaa/SZPifC7BkcraZVejAe8=
+github.com/rogpeppe/go-internal v1.9.0/go.mod h1:WtVeX8xhTBvf0smdhujwtBcq4Qrzq/fJaraNFVN+nFs=
+github.com/sirupsen/logrus v1.9.3 h1:dueUQJ1C2q9oE3F7wvmSGAaVtTmUizReu6fjN8uqzbQ=
+github.com/sirupsen/logrus v1.9.3/go.mod h1:naHLuLoDiP4jHNo9R0sCBMtWGeIprob74mVsIT4qYEQ=
+github.com/stretchr/objx v0.1.0/go.mod h1:HFkY916IF+rwdDfMAkV7OtwuqBVzrE8GR6GFx+wExME=
+github.com/stretchr/testify v1.7.0/go.mod h1:6Fq8oRcR53rry900zMqJjRRixrwX3KX962/h/Wwjteg=
+github.com/stretchr/testify v1.8.1 h1:w7B6lhMri9wdJUVmEZPGGhZzrYTPvgJArz7wNPgYKsk=
+github.com/stretchr/testify v1.8.1/go.mod h1:w2LPCIKwWwSfY2zedu0+kehJoqGctiVI29o6fzry7u4=
+golang.org/x/crypto v0.7.0 h1:AvwMYaRytfdeVt3u6mLaxYtErKYjxA2OXjJ1HHq6t3A=
+golang.org/x/crypto v0.7.0/go.mod h1:pYwdfH91IfpZVANVyUOhSIPZaFoJGxTFbZhFTx+dXZU=
+golang.org/x/sys v0.0.0-20220715151400-c0bba94af5f8/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
+golang.org/x/sys v0.6.0 h1:MVltZSvRTcU2ljQOhs94SXPftV6DCNnZViHeQps87pQ=
+golang.org/x/sys v0.6.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
+golang.org/x/sys v0.21.0 h1:rF+pYz3DAGSQAxAu1CbC7catZg4ebC4UIeIhKxBZvws=
+golang.org/x/sys v0.21.0/go.mod h1:/VUhepiaJMQUp4+oa/7Zr1D23ma6VTLIYjOOTFZPUcA=
+gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0=
+gopkg.in/yaml.v3 v3.0.0-20200313102051-9f266ea9e77c/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM=
+gopkg.in/yaml.v3 v3.0.1 h1:fxVm/GzAzEWqLHuvctI91KS9hhNmmWOoWu0XTYJS7CA=
+gopkg.in/yaml.v3 v3.0.1/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM=

monitor.go

@@ -0,0 +1,217 @@
+package main
+
+import (
+	"context"
+	"errors"
+	"sync/atomic"
+	"time"
+
+	tz "github.com/ecadlabs/gotez/v2"
+	"github.com/ecadlabs/gotez/v2/client"
+	"github.com/ecadlabs/gotez/v2/protocol/core"
+	log "github.com/sirupsen/logrus"
+)
+
+const (
+	defaultTimeout        = 30 * time.Second
+	defaultTolerance      = 1 * time.Second
+	defaultReconnectDelay = 10 * time.Second
+)
+
+type HeadMonitor struct {
+	Client         *client.Client
+	ChainID        *tz.ChainID
+	Timeout        time.Duration
+	Tolerance      time.Duration
+	ReconnectDelay time.Duration
+	UseTimestamps  bool
+
+	status atomic.Bool
+	cancel context.CancelFunc
+	done   chan struct{}
+}
+
+func (h *HeadMonitor) Status() bool {
+	return h.status.Load()
+}
+
+func (h *HeadMonitor) context(ctx context.Context) (context.Context, context.CancelFunc) {
+	return context.WithTimeout(ctx, h.timeout())
+}
+
+func (h *HeadMonitor) timeout() time.Duration {
+	if h.Timeout != 0 {
+		return h.Timeout
+	}
+	return defaultTimeout
+}
+
+func (h *HeadMonitor) tolerance() time.Duration {
+	if h.Tolerance != 0 {
+		return h.Tolerance
+	}
+	return defaultTolerance
+}
+
+func (h *HeadMonitor) reconnectDelay() time.Duration {
+	if h.ReconnectDelay != 0 {
+		return h.ReconnectDelay
+	}
+	return defaultReconnectDelay
+}
+
+func (h *HeadMonitor) getMinBlockDelay(c context.Context, block string, protocol *tz.ProtocolHash) (time.Duration, error) {
+	ctx, cancel := h.context(c)
+	defer cancel()
+	consts, err := h.Client.Constants(ctx, &client.ContextRequest{
+		Chain:    h.ChainID.String(),
+		Block:    block,
+		Protocol: protocol,
+	})
+	if err != nil {
+		return 0, err
+	}
+	delay := time.Duration(consts.GetMinimalBlockDelay()) * time.Second
+	log.Debugf("%s delay = %v", block, delay)
+	return delay, nil
+}
+
+func (h *HeadMonitor) getShellHeader(c context.Context, block *tz.BlockHash) (*core.ShellHeader, error) {
+	ctx, cancel := h.context(c)
+	defer cancel()
+	return h.Client.BlockShellHeader(ctx, &client.SimpleRequest{
+		Chain: h.ChainID.String(),
+		Block: block.String(),
+	})
+}
+
+func (h *HeadMonitor) getBlockInfo(c context.Context, block string) (*client.BasicBlockInfo, error) {
+	ctx, cancel := h.context(c)
+	defer cancel()
+	return h.Client.BasicBlockInfo(ctx, h.ChainID.String(), block)
+}
+
+func (h *HeadMonitor) Start() {
+	ctx, cancel := context.WithCancel(context.Background())
+	h.cancel = cancel
+	h.done = make(chan struct{})
+	go h.serve(ctx)
+}
+
+func (h *HeadMonitor) Stop(ctx context.Context) error {
+	h.cancel()
+	select {
+	case <-h.done:
+		return nil
+	case <-ctx.Done():
+		return ctx.Err()
+	}
+}
+
+func (h *HeadMonitor) serve(ctx context.Context) {
+	defer close(h.done)
+	var err error
+	for {
+		h.status.Store(false)
+		if err != nil {
+			log.Error(err)
+			t := time.After(h.reconnectDelay())
+			select {
+			case <-t:
+			case <-ctx.Done():
+				return
+			}
+		}
+
+		var bi *client.BasicBlockInfo
+		bi, err = h.getBlockInfo(ctx, "head")
+		if err != nil {
+			if errors.Is(err, context.Canceled) {
+				return
+			}
+			continue
+		}
+		var sh *core.ShellHeader
+		sh, err = h.getShellHeader(ctx, bi.Hash)
+		if err != nil {
+			if errors.Is(err, context.Canceled) {
+				return
+			}
+			continue
+		}
+		var timestamp time.Time
+		if h.UseTimestamps {
+			timestamp = sh.Timestamp.Time()
+		} else {
+			timestamp = time.Now()
+		}
+
+		protoNum := sh.Proto
+		var minBlockDelay time.Duration
+		minBlockDelay, err = h.getMinBlockDelay(ctx, bi.Hash.String(), bi.Protocol)
+		if err != nil {
+			if errors.Is(err, context.Canceled) {
+				return
+			}
+			continue
+		}
+		var (
+			stream <-chan *client.Head
+			errCh  <-chan error
+		)
+		stream, errCh, err = h.Client.Heads(ctx, &client.HeadsRequest{Chain: h.ChainID.String()})
+		if err != nil {
+			if errors.Is(err, context.Canceled) {
+				return
+			}
+			continue
+		}
+
+	Recv:
+		for {
+			select {
+			case err = <-errCh:
+				if errors.Is(err, context.Canceled) {
+					return
+				}
+				break Recv
+
+			case head := <-stream:
+				var t time.Time
+				if h.UseTimestamps {
+					t = head.Timestamp.Time()
+				} else {
+					t = time.Now()
+				}
+				status := t.Before(timestamp.Add(minBlockDelay + h.tolerance()))
+				log.Debugf("%v: %t", t, status)
+				h.status.Store(status)
+				timestamp = t
+				if head.Proto == protoNum {
+					break
+				}
+
+				// update constant
+				var proto *core.BlockProtocols
+				proto, err = h.Client.BlockProtocols(ctx, &client.SimpleRequest{
+					Chain: h.ChainID.String(),
+					Block: head.Hash.String(),
+				})
+				if err != nil {
+					if errors.Is(err, context.Canceled) {
+						return
+					}
+					break Recv
+				}
+				minBlockDelay, err = h.getMinBlockDelay(ctx, head.Hash.String(), proto.Protocol)
+				if err != nil {
+					if errors.Is(err, context.Canceled) {
+						return
+					}
+					break Recv
+				}
+				protoNum = head.Proto
+			}
+		}
+	}
+}