go-proxy/splitter/simple/impl_test.go

package simple

import (
	"bufio"
	"encoding/json"
	"io"
	"net"
	"sync"
	"testing"
	"time"

	"dappco.re/go/proxy"
	"dappco.re/go/proxy/pool"
)

type activeStrategy struct{}

func (a activeStrategy) Connect()                                    {}
func (a activeStrategy) Submit(string, string, string, string) int64 { return 0 }
func (a activeStrategy) Disconnect()                                 {}
func (a activeStrategy) IsActive() bool                              { return true }

func TestSimpleSplitter_OnLogin_Good(t *testing.T) {
	splitter := NewSimpleSplitter(&proxy.Config{ReuseTimeout: 30}, nil, func(listener pool.StratumListener) pool.Strategy {
		return activeStrategy{}
	})
	miner := &proxy.Miner{}
	job := proxy.Job{JobID: "job-1", Blob: "blob"}
	mapper := &SimpleMapper{
		id:         7,
		strategy:   activeStrategy{},
		currentJob: job,
		idleAt:     time.Now(),
	}
	splitter.idle[mapper.id] = mapper

	splitter.OnLogin(&proxy.LoginEvent{Miner: miner})

	if miner.RouteID() != mapper.id {
		t.Fatalf("expected reclaimed mapper route id %d, got %d", mapper.id, miner.RouteID())
	}
	if got := miner.CurrentJob().JobID; got != job.JobID {
		t.Fatalf("expected current job to be restored on reuse, got %q", got)
	}
}

func TestSimpleSplitter_OnLogin_Ugly(t *testing.T) {
	splitter := NewSimpleSplitter(&proxy.Config{ReuseTimeout: 30}, nil, func(listener pool.StratumListener) pool.Strategy {
		return activeStrategy{}
	})
	miner := &proxy.Miner{}
	expired := &SimpleMapper{
		id:       7,
		strategy: activeStrategy{},
		idleAt:   time.Now().Add(-time.Minute),
	}
	splitter.idle[expired.id] = expired

	splitter.OnLogin(&proxy.LoginEvent{Miner: miner})

	if miner.RouteID() == expired.id {
		t.Fatalf("expected expired mapper not to be reclaimed")
	}
	if miner.RouteID() != 0 {
		t.Fatalf("expected a new mapper to be allocated, got route id %d", miner.RouteID())
	}
	if len(splitter.active) != 1 {
		t.Fatalf("expected one active mapper, got %d", len(splitter.active))
	}
	if len(splitter.idle) != 1 {
		t.Fatalf("expected expired mapper to remain idle until GC, got %d idle mappers", len(splitter.idle))
	}
}

func TestSimpleSplitter_Upstreams_Good(t *testing.T) {
	splitter := NewSimpleSplitter(&proxy.Config{ReuseTimeout: 30}, nil, func(listener pool.StratumListener) pool.Strategy {
		return activeStrategy{}
	})
	splitter.active[1] = &SimpleMapper{id: 1, strategy: activeStrategy{}}
	splitter.idle[2] = &SimpleMapper{id: 2, strategy: activeStrategy{}, idleAt: time.Now()}

	stats := splitter.Upstreams()

	if stats.Active != 1 {
		t.Fatalf("expected one active upstream, got %d", stats.Active)
	}
	if stats.Sleep != 1 {
		t.Fatalf("expected one sleeping upstream, got %d", stats.Sleep)
	}
	if stats.Error != 0 {
		t.Fatalf("expected no error upstreams, got %d", stats.Error)
	}
	if stats.Total != 2 {
		t.Fatalf("expected total upstreams to be 2, got %d", stats.Total)
	}
}

func TestSimpleSplitter_Upstreams_Ugly(t *testing.T) {
	splitter := NewSimpleSplitter(&proxy.Config{ReuseTimeout: 30}, nil, func(listener pool.StratumListener) pool.Strategy {
		return activeStrategy{}
	})
	splitter.active[1] = &SimpleMapper{id: 1, strategy: activeStrategy{}, stopped: true}
	splitter.idle[2] = &SimpleMapper{id: 2, strategy: activeStrategy{}, stopped: true, idleAt: time.Now()}

	stats := splitter.Upstreams()

	if stats.Active != 0 {
		t.Fatalf("expected no active upstreams, got %d", stats.Active)
	}
	if stats.Sleep != 0 {
		t.Fatalf("expected no sleeping upstreams, got %d", stats.Sleep)
	}
	if stats.Error != 2 {
		t.Fatalf("expected both upstreams to be counted as error, got %d", stats.Error)
	}
	if stats.Total != 2 {
		t.Fatalf("expected total upstreams to be 2, got %d", stats.Total)
	}
}

type discardConn struct{}

func (discardConn) Read([]byte) (int, error)         { return 0, io.EOF }
func (discardConn) Write(p []byte) (int, error)      { return len(p), nil }
func (discardConn) Close() error                     { return nil }
func (discardConn) LocalAddr() net.Addr              { return nil }
func (discardConn) RemoteAddr() net.Addr             { return nil }
func (discardConn) SetDeadline(time.Time) error      { return nil }
func (discardConn) SetReadDeadline(time.Time) error  { return nil }
func (discardConn) SetWriteDeadline(time.Time) error { return nil }

func TestSimpleMapper_OnResultAccepted_Expired(t *testing.T) {
	bus := proxy.NewEventBus()
	events := make(chan proxy.Event, 1)
	var once sync.Once
	bus.Subscribe(proxy.EventAccept, func(e proxy.Event) {
		once.Do(func() {
			events <- e
		})
	})

	miner := proxy.NewMiner(discardConn{}, 3333, nil)
	miner.SetID(1)
	mapper := &SimpleMapper{
		miner:      miner,
		currentJob: proxy.Job{JobID: "job-new", Blob: "blob-new", Target: "b88d0600"},
		prevJob:    proxy.Job{JobID: "job-old", Blob: "blob-old", Target: "b88d0600"},
		events:     bus,
		pending: map[int64]submitContext{
			7: {RequestID: 9, StartedAt: time.Now(), JobID: "job-old"},
		},
	}

	mapper.OnResultAccepted(7, true, "")

	select {
	case event := <-events:
		if !event.Expired {
			t.Fatalf("expected expired share to be flagged")
		}
		if event.Job == nil || event.Job.JobID != "job-old" {
			t.Fatalf("expected previous job to be attached, got %+v", event.Job)
		}
	case <-time.After(time.Second):
		t.Fatal("expected accept event")
	}
}

func TestSimpleMapper_Submit_InvalidJob_Good(t *testing.T) {
	minerConn, clientConn := net.Pipe()
	defer minerConn.Close()
	defer clientConn.Close()

	miner := proxy.NewMiner(minerConn, 3333, nil)
	mapper := &SimpleMapper{
		miner:      miner,
		currentJob: proxy.Job{JobID: "job-1", Blob: "blob", Target: "b88d0600"},
		prevJob:    proxy.Job{JobID: "job-0", Blob: "blob", Target: "b88d0600"},
		strategy:   activeStrategy{},
		pending:    make(map[int64]submitContext),
	}

	done := make(chan struct{})
	go func() {
		mapper.Submit(&proxy.SubmitEvent{
			Miner:     miner,
			JobID:     "job-missing",
			Nonce:     "deadbeef",
			Result:    "hash",
			RequestID: 9,
		})
		close(done)
	}()

	line, err := bufio.NewReader(clientConn).ReadBytes('\n')
	if err != nil {
		t.Fatalf("read error reply: %v", err)
	}
	<-done

	var payload struct {
		ID    float64 `json:"id"`
		Error struct {
			Message string `json:"message"`
		} `json:"error"`
	}
	if err := json.Unmarshal(line, &payload); err != nil {
		t.Fatalf("unmarshal error reply: %v", err)
	}
	if payload.ID != 9 {
		t.Fatalf("expected request id 9, got %v", payload.ID)
	}
	if payload.Error.Message != "Invalid job id" {
		t.Fatalf("expected invalid job error, got %q", payload.Error.Message)
	}
	if len(mapper.pending) != 0 {
		t.Fatalf("expected invalid submit not to create a pending entry")
	}
}