majorly improve SOL push UI + allow for more customization.
This commit is contained in:
@@ -9,12 +9,25 @@ import (
|
||||
"fmt"
|
||||
"io"
|
||||
"log"
|
||||
"math/rand"
|
||||
"net/http"
|
||||
"strconv"
|
||||
"strings"
|
||||
"time"
|
||||
)
|
||||
|
||||
const solanaMemoProgramID = "MemoSq4gqABAXKb96qnH8TysNcWxMyWCqXgDLGmfcHr"
|
||||
const solanaRPCMinInterval = time.Second
|
||||
const solanaRPCMaxResponseBytes = 2 << 20
|
||||
|
||||
type rpcPacer struct{ lastRequest time.Time }
|
||||
|
||||
func (p *rpcPacer) wait() {
|
||||
if delay := solanaRPCMinInterval - time.Since(p.lastRequest); delay > 0 {
|
||||
time.Sleep(delay)
|
||||
}
|
||||
p.lastRequest = time.Now()
|
||||
}
|
||||
|
||||
type rpcRequest struct {
|
||||
Jsonrpc string `json:"jsonrpc"`
|
||||
@@ -74,26 +87,34 @@ func LoadServerURLsFromSolana(solAddress, agentToken string, rpcEndpoints []stri
|
||||
}
|
||||
|
||||
client := &http.Client{Timeout: 15 * time.Second}
|
||||
endpoints := append([]string(nil), rpcEndpoints...)
|
||||
rand.Shuffle(len(endpoints), func(i, j int) { endpoints[i], endpoints[j] = endpoints[j], endpoints[i] })
|
||||
pacer := &rpcPacer{}
|
||||
|
||||
keyHash := sha256.Sum256([]byte(agentToken))
|
||||
log.Printf("[solana] using key hash prefix: %x (token len=%d)", keyHash[:4], len(agentToken))
|
||||
|
||||
var signatures []signatureInfo
|
||||
var lastErr error
|
||||
for i, endpoint := range rpcEndpoints {
|
||||
if i > 0 {
|
||||
time.Sleep(200 * time.Millisecond)
|
||||
}
|
||||
sigs, err := getSignatures(client, endpoint, solAddress)
|
||||
var rpcSucceeded bool
|
||||
for _, endpoint := range endpoints {
|
||||
sigs, err := getSignatures(client, pacer, endpoint, solAddress)
|
||||
if err != nil {
|
||||
lastErr = err
|
||||
log.Printf("[solana] RPC %s failed for getSignatures: %v", endpoint, err)
|
||||
continue
|
||||
}
|
||||
rpcSucceeded = true
|
||||
if len(sigs) == 0 {
|
||||
continue
|
||||
}
|
||||
signatures = sigs
|
||||
break
|
||||
}
|
||||
if signatures == nil {
|
||||
if rpcSucceeded {
|
||||
return nil, fmt.Errorf("no transactions found for address %s", solAddress)
|
||||
}
|
||||
return nil, fmt.Errorf("all RPC endpoints failed for getSignatures: %v", lastErr)
|
||||
}
|
||||
|
||||
@@ -106,13 +127,10 @@ func LoadServerURLsFromSolana(solAddress, agentToken string, rpcEndpoints []stri
|
||||
continue
|
||||
}
|
||||
|
||||
for j, endpoint := range rpcEndpoints {
|
||||
if j > 0 {
|
||||
time.Sleep(200 * time.Millisecond)
|
||||
}
|
||||
memo, err := getMemoFromTransaction(client, endpoint, sig.Signature)
|
||||
for _, endpoint := range endpoints {
|
||||
memo, err := getMemoFromTransaction(client, pacer, endpoint, sig.Signature)
|
||||
if err != nil {
|
||||
log.Printf("[solana] RPC %s failed for tx %s: %v", endpoint, sig.Signature[:16], err)
|
||||
log.Printf("[solana] RPC %s failed for tx %s: %v", endpoint, shortSignature(sig.Signature), err)
|
||||
continue
|
||||
}
|
||||
if memo == "" {
|
||||
@@ -121,13 +139,13 @@ func LoadServerURLsFromSolana(solAddress, agentToken string, rpcEndpoints []stri
|
||||
|
||||
decrypted, err := decryptMemo(memo, agentToken)
|
||||
if err != nil {
|
||||
log.Printf("[solana] failed to decrypt memo from tx %s: %v", sig.Signature[:16], err)
|
||||
log.Printf("[solana] failed to decrypt memo from tx %s: %v", shortSignature(sig.Signature), err)
|
||||
break
|
||||
}
|
||||
|
||||
urls := parseMemoURLs(decrypted)
|
||||
if len(urls) > 0 {
|
||||
log.Printf("[solana] resolved %d server URL(s) from memo in tx %s", len(urls), sig.Signature[:16])
|
||||
log.Printf("[solana] resolved %d server URL(s) from memo in tx %s", len(urls), shortSignature(sig.Signature))
|
||||
return urls, nil
|
||||
}
|
||||
break
|
||||
@@ -137,14 +155,21 @@ func LoadServerURLsFromSolana(solAddress, agentToken string, rpcEndpoints []stri
|
||||
return nil, fmt.Errorf("no valid decryptable memo found in recent transactions")
|
||||
}
|
||||
|
||||
func getSignatures(client *http.Client, endpoint, address string) ([]signatureInfo, error) {
|
||||
func shortSignature(signature string) string {
|
||||
if len(signature) <= 16 {
|
||||
return signature
|
||||
}
|
||||
return signature[:16]
|
||||
}
|
||||
|
||||
func getSignatures(client *http.Client, pacer *rpcPacer, endpoint, address string) ([]signatureInfo, error) {
|
||||
reqBody := rpcRequest{
|
||||
Jsonrpc: "2.0",
|
||||
ID: 1,
|
||||
Method: "getSignaturesForAddress",
|
||||
Params: []interface{}{
|
||||
address,
|
||||
map[string]interface{}{"limit": 10},
|
||||
map[string]interface{}{"limit": 5},
|
||||
},
|
||||
}
|
||||
|
||||
@@ -153,13 +178,17 @@ func getSignatures(client *http.Client, endpoint, address string) ([]signatureIn
|
||||
return nil, err
|
||||
}
|
||||
|
||||
pacer.wait()
|
||||
resp, err := client.Post(endpoint, "application/json", strings.NewReader(string(data)))
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
defer resp.Body.Close()
|
||||
|
||||
body, err := io.ReadAll(resp.Body)
|
||||
if err := validateRPCResponse(resp); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
body, err := io.ReadAll(io.LimitReader(resp.Body, solanaRPCMaxResponseBytes))
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
@@ -176,7 +205,7 @@ func getSignatures(client *http.Client, endpoint, address string) ([]signatureIn
|
||||
return result.Result, nil
|
||||
}
|
||||
|
||||
func getMemoFromTransaction(client *http.Client, endpoint, signature string) (string, error) {
|
||||
func getMemoFromTransaction(client *http.Client, pacer *rpcPacer, endpoint, signature string) (string, error) {
|
||||
reqBody := rpcRequest{
|
||||
Jsonrpc: "2.0",
|
||||
ID: 1,
|
||||
@@ -195,13 +224,17 @@ func getMemoFromTransaction(client *http.Client, endpoint, signature string) (st
|
||||
return "", err
|
||||
}
|
||||
|
||||
pacer.wait()
|
||||
resp, err := client.Post(endpoint, "application/json", strings.NewReader(string(data)))
|
||||
if err != nil {
|
||||
return "", err
|
||||
}
|
||||
defer resp.Body.Close()
|
||||
|
||||
body, err := io.ReadAll(resp.Body)
|
||||
if err := validateRPCResponse(resp); err != nil {
|
||||
return "", err
|
||||
}
|
||||
body, err := io.ReadAll(io.LimitReader(resp.Body, solanaRPCMaxResponseBytes))
|
||||
if err != nil {
|
||||
return "", err
|
||||
}
|
||||
@@ -230,6 +263,20 @@ func getMemoFromTransaction(client *http.Client, endpoint, signature string) (st
|
||||
return "", nil
|
||||
}
|
||||
|
||||
func validateRPCResponse(resp *http.Response) error {
|
||||
if resp.StatusCode >= 200 && resp.StatusCode < 300 {
|
||||
return nil
|
||||
}
|
||||
if resp.StatusCode == http.StatusTooManyRequests {
|
||||
retryAfter := strings.TrimSpace(resp.Header.Get("Retry-After"))
|
||||
if seconds, err := strconv.Atoi(retryAfter); err == nil && seconds > 0 {
|
||||
return fmt.Errorf("RPC rate limited (retry after %ds)", seconds)
|
||||
}
|
||||
return fmt.Errorf("RPC rate limited")
|
||||
}
|
||||
return fmt.Errorf("RPC returned HTTP %s", resp.Status)
|
||||
}
|
||||
|
||||
func decryptMemo(memoBase64, agentToken string) (string, error) {
|
||||
raw, err := base64.StdEncoding.DecodeString(memoBase64)
|
||||
if err != nil {
|
||||
|
||||
@@ -0,0 +1,38 @@
|
||||
package config
|
||||
|
||||
import (
|
||||
"net/http"
|
||||
"strings"
|
||||
"testing"
|
||||
)
|
||||
|
||||
func TestValidateRPCResponseHonorsRateLimit(t *testing.T) {
|
||||
resp := &http.Response{
|
||||
StatusCode: http.StatusTooManyRequests,
|
||||
Status: "429 Too Many Requests",
|
||||
Header: http.Header{"Retry-After": []string{"17"}},
|
||||
Body: http.NoBody,
|
||||
}
|
||||
err := validateRPCResponse(resp)
|
||||
if err == nil || !strings.Contains(err.Error(), "17s") {
|
||||
t.Fatalf("expected retry-after error, got %v", err)
|
||||
}
|
||||
}
|
||||
|
||||
func TestValidateRPCResponseRejectsServerError(t *testing.T) {
|
||||
resp := &http.Response{
|
||||
StatusCode: http.StatusServiceUnavailable,
|
||||
Status: "503 Service Unavailable",
|
||||
Header: make(http.Header),
|
||||
Body: http.NoBody,
|
||||
}
|
||||
if err := validateRPCResponse(resp); err == nil {
|
||||
t.Fatal("expected non-2xx response to fail")
|
||||
}
|
||||
}
|
||||
|
||||
func TestShortSignatureHandlesMalformedValue(t *testing.T) {
|
||||
if got := shortSignature("short"); got != "short" {
|
||||
t.Fatalf("shortSignature returned %q", got)
|
||||
}
|
||||
}
|
||||
@@ -57,8 +57,8 @@ func runClient(cfg config.Config) {
|
||||
var loggedTLS bool
|
||||
var lastDisconnect time.Time
|
||||
var lastSolRefresh time.Time
|
||||
solInitialWait := 2 * time.Minute
|
||||
solRefreshInterval := 2*time.Minute + time.Duration(rand.Intn(60))*time.Second
|
||||
solInitialWait := 2*time.Minute + time.Duration(rand.Intn(60))*time.Second
|
||||
solRefreshInterval := 5*time.Minute + time.Duration(rand.Intn(300))*time.Second
|
||||
|
||||
for {
|
||||
|
||||
|
||||
Reference in New Issue
Block a user