core/go-proxy
8
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

fix(pool): handle upstream error responses

Browse source 
Co-Authored-By: Virgil <virgil@lethean.io>
This commit is contained in:
Virgil 2026-04-04 17:07:26 +00:00
parent 23623a97d3
commit 1d6176153c
2 changed files with 121 additions and 28 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

65
pool/client.go
View file

@ -235,34 +235,6 @@ func (c *StratumClient) notifyDisconnect() {
}
func (c *StratumClient) handleMessage(response jsonRPCResponse) {
if len(response.Result) > 0 {
var loginResult struct {
ID string `json:"id"`
Job proxy.Job `json:"job"`
}
if json.Unmarshal(response.Result, &loginResult) == nil && loginResult.ID != "" {
c.sessionID = loginResult.ID
if loginResult.Job.IsValid() {
loginResult.Job.ClientID = c.sessionID
c.active = true
if c.listener != nil {
c.listener.OnJob(loginResult.Job)
}
}
return
}
if c.listener != nil {
accepted := response.Error == nil
errorMessage := ""
if response.Error != nil {
errorMessage = response.Error.Message
}
c.listener.OnResultAccepted(response.ID, accepted, errorMessage)
}
return
}
if response.Method == "job" {
var payload proxy.Job
if json.Unmarshal(response.Params, &payload) == nil && payload.IsValid() {
@ -272,5 +244,42 @@ func (c *StratumClient) handleMessage(response jsonRPCResponse) {
c.listener.OnJob(payload)
}
}
return
}
if response.ID == 1 && c.sessionID == "" {
if len(response.Result) > 0 {
var loginResult struct {
ID string `json:"id"`
Job proxy.Job `json:"job"`
}
if json.Unmarshal(response.Result, &loginResult) == nil && loginResult.ID != "" {
c.sessionID = loginResult.ID
if loginResult.Job.IsValid() {
loginResult.Job.ClientID = c.sessionID
c.active = true
if c.listener != nil {
c.listener.OnJob(loginResult.Job)
}
}
}
}
if response.Error != nil {
c.Disconnect()
}
return
}
if response.ID == 0 || c.listener == nil {
return
}
accepted := response.Error == nil
errorMessage := ""
if response.Error != nil {
errorMessage = response.Error.Message
}
if !accepted || len(response.Result) > 0 {
c.listener.OnResultAccepted(response.ID, accepted, errorMessage)
}
}

84
pool/client_test.go
View file

@ -87,3 +87,87 @@ func TestStratumClient_Disconnect_Good(t *testing.T) {
t.Fatalf("expected one disconnect callback, got %d", got)
}
}
type resultCapturingListener struct {
mu sync.Mutex
sequence int64
accepted bool
errorMessage string
results int
disconnects int
}
func (l *resultCapturingListener) OnJob(job proxy.Job) {}
func (l *resultCapturingListener) OnResultAccepted(sequence int64, accepted bool, errorMessage string) {
l.mu.Lock()
l.sequence = sequence
l.accepted = accepted
l.errorMessage = errorMessage
l.results++
l.mu.Unlock()
}
func (l *resultCapturingListener) OnDisconnect() {
l.mu.Lock()
l.disconnects++
l.mu.Unlock()
}
func (l *resultCapturingListener) Snapshot() (int64, bool, string, int, int) {
l.mu.Lock()
defer l.mu.Unlock()
return l.sequence, l.accepted, l.errorMessage, l.results, l.disconnects
}
func TestStratumClient_HandleMessage_Bad(t *testing.T) {
listener := &resultCapturingListener{}
client := &StratumClient{
listener: listener,
sessionID: "session-1",
}
client.handleMessage(jsonRPCResponse{
ID: 7,
Error: &jsonRPCErrorBody{
Code: -1,
Message: "Low difficulty share",
},
})
sequence, accepted, errorMessage, results, disconnects := listener.Snapshot()
if sequence != 7 || accepted || errorMessage != "Low difficulty share" || results != 1 {
t.Fatalf("expected rejected submit callback, got sequence=%d accepted=%v error=%q results=%d", sequence, accepted, errorMessage, results)
}
if disconnects != 0 {
t.Fatalf("expected no disconnect on submit rejection, got %d", disconnects)
}
}
func TestStratumClient_HandleMessage_Ugly(t *testing.T) {
serverConn, clientConn := net.Pipe()
defer clientConn.Close()
listener := &resultCapturingListener{}
client := &StratumClient{
listener: listener,
conn: serverConn,
}
defer client.Disconnect()
client.handleMessage(jsonRPCResponse{
ID: 1,
Error: &jsonRPCErrorBody{
Code: -1,
Message: "Unauthenticated",
},
})
_, _, _, results, disconnects := listener.Snapshot()
if results != 0 {
t.Fatalf("expected login rejection not to be reported as a share result, got %d results", results)
}
if disconnects != 1 {
t.Fatalf("expected login rejection to disconnect once, got %d", disconnects)
}
}