bothub.ru

js 171225

// test_abelian29_final.js
// Final Abelian Stratum proxy (forward / local test mode)
// Usage:
//   Forward (default): node test_abelian29_final.js
//   Local test accept: MODE=local node test_abelian29_final.js

const net = require('net');
const http = require('http');

const MODE = (process.env.MODE || process.env.NODE_ENV || 'forward').toLowerCase(); // 'forward' or 'local'

const STRATUM_HOST = '0.0.0.0';
const STRATUM_PORT = 3333;

const RPC_HOST = '127.0.0.1';
const RPC_PORT = 8668;
const RPC_USER = 'tteFTlJ7YOfGDA2KBMHKqnDnXeE=';
const RPC_PASS = 'SOkvF8sxay8ViOxpgbraHmqJmSU=';

const JOB_POLL_INTERVAL = 8000;
const STATS_PRINT_INTERVAL = 30000;

let globalJob = null;
const clients = new Set();
const stats = { workers: {}, totalAccepted: 0, totalRejected: 0, totalLocalAccepted: 0 };

// helpers
function sendLine(sock, obj) {
  const s = JSON.stringify(obj);
  try { sock.write(s + '\n'); } catch (e) {}
  console.log('=>', s);
}

function httpRpcRaw(payload) {
  const options = {
    hostname: RPC_HOST, port: RPC_PORT, path: '/', method: 'POST',
    headers: { 'Content-Type': 'application/json', 'Content-Length': Buffer.byteLength(payload) },
    auth: `${RPC_USER}:${RPC_PASS}`
  };
  return new Promise((resolve, reject) => {
    const req = http.request(options, (res) => {
      let data = '';
      res.setEncoding('utf8');
      res.on('data', (c) => data += c);
      res.on('end', () => resolve({ raw: data, status: res.statusCode }));
    });
    req.on('error', (e) => reject(e));
    req.write(payload); req.end();
  });
}

async function rpcCall(method, params = []) {
  const payload = JSON.stringify({ jsonrpc: "1.0", id: "proxy", method, params });
  try {
    const r = await httpRpcRaw(payload);
    try { return { parsed: JSON.parse(r.raw), raw: r.raw }; } catch (e) { return { parsed: null, raw: r.raw }; }
  } catch (e) { return { parsed: null, raw: null, error: e.message || String(e) }; }
}

function strip0x(s) { return s ? String(s).replace(/^0x/, '') : ''; }
function padEven(s) { let h = String(s||'').replace(/^0x/,''); if (h.length % 2) h = '0' + h; return h; }
function revHex(s) { let h = padEven(s), out=''; for (let i=0;i<h.length;i+=2) out = h.substr(i,2) + out; return out; }

function recordShare(worker, accepted, local=false) {
  if (!worker) worker = 'unknown';
  if (!stats.workers[worker]) stats.workers[worker] = { accepted:0, rejected:0, local:0, last:0 };
  if (accepted) { stats.workers[worker].accepted++; stats.totalAccepted++; } else { stats.workers[worker].rejected++; stats.totalRejected++; }
  if (local && accepted) { stats.workers[worker].local++; stats.totalLocalAccepted++; }
  stats.workers[worker].last = Date.now();
}

function printStats() {
  console.log('=== PROXY STATS ===');
  console.log(`totalAccepted=${stats.totalAccepted} totalRejected=${stats.totalRejected} totalLocalAccepted=${stats.totalLocalAccepted}`);
  for (const w of Object.keys(stats.workers)) {
    const s = stats.workers[w];
    console.log(`worker=${w} a=${s.accepted} r=${s.rejected} local=${s.local} last=${s.last ? new Date(s.last).toISOString() : '-'}`);
  }
  console.log('===================');
}

// fetch job from node
async function fetchJob() {
  const r = await rpcCall('getwork', [null]);
  console.log('getwork([null]) =>', (r.raw || '').slice(0,900));
  if (r.parsed && r.parsed.result) {
    const res = r.parsed.result;
    return {
      jobid: res.jobid || '',
      contenthash: strip0x(res.contenthash || res.contentHash || ''),
      epochseed: strip0x(res.epochseed || res.seed || ''),
      target: strip0x(res.targetboundary || res.target || ''),
      extranonce: res.extranonce,
      extranoncebitsnum: res.extranoncebitsnum,
      raw: res
    };
  }
  return null;
}

// broadcast object-style notify expected by TeamRedMiner
function broadcastJobToSocket(sock, job) {
  try {
    const setObj = {
      id: null, method: 'mining.set', params: {
        epoch: job.raw && job.raw.epoch !== undefined ? job.raw.epoch.toString(16) : '',
        target: job.target || '',
        algo: 'abelethash',
        extranonce: job.extranonce !== undefined ? padEven(String(job.extranonce.toString(16))) : '',
        extra_nonce_bits_num: job.extranoncebitsnum !== undefined ? String(job.extranoncebitsnum) : ''
      }
    };
    sendLine(sock, setObj);

    const jobShort = (job.jobid || '').substring(0,8);
    const notifyObj = {
      id: null, method: 'mining.notify', params: {
        job_id: jobShort,
        height: job.raw && job.raw.height ? job.raw.height.toString(16) : '',
        content_hash: job.contenthash || '',
        clean_job: '0'
      }
    };
    sendLine(sock, notifyObj);
  } catch (e) {
    console.warn('broadcastJobToSocket error', e && e.message ? e.message : e);
  }
}

function broadcastJob(job) {
  for (const s of clients) broadcastJobToSocket(s, job);
}

// stratum server
const server = net.createServer((sock) => {
  console.log('Client connected', sock.remoteAddress, sock.remotePort);
  clients.add(sock);
  const state = { buf: '', extraNonce1: padEven(Math.floor(Math.random()*0xffffffff).toString(16).padStart(8,'0')), authorized: false };

  sock.setEncoding('utf8');
  sock.on('data', async (data) => {
    state.buf += data;
    let idx;
    while ((idx = state.buf.indexOf('\n')) !== -1) {
      const line = state.buf.slice(0, idx).trim();
      state.buf = state.buf.slice(idx + 1);
      if (!line) continue;
      console.log('IN:', line);
      let msg;
      try { msg = JSON.parse(line); } catch (e) { console.error('parse error', e.message); continue; }
      await handleClientMessage(sock, state, msg);
    }
  });

  sock.on('close', () => { clients.delete(sock); console.log('client closed'); });
  sock.on('error', (e) => { clients.delete(sock); console.error('sock err', e && e.message ? e.message : e); });
});

async function handleClientMessage(sock, state, msg) {
  const method = msg.method;
  if (!method) return;

  if (method === 'mining.subscribe') {
    const reply = [ [[state.extraNonce1, 8]], 'proxy:abel:1.0' ];
    sendLine(sock, { id: msg.id || null, result: reply, error: null });
    if (globalJob) broadcastJobToSocket(sock, globalJob);
    return;
  }

  if (method === 'mining.authorize') {
    state.authorized = true;
    sendLine(sock, { id: msg.id || null, result: true, error: null });
    return;
  }

  if (method === 'mining.submit') {
    const params = msg.params || [];
    console.log('mining.submit params:', params);
    const worker = params[0] || 'unknown';
    if (params.length < 1) { sendLine(sock, { id: msg.id || null, result: false, error: 'invalid params' }); recordShare(worker, false); return; }

    let last = params[params.length - 1];
    if (typeof last !== 'string') last = String(last);
    let nonce = padEven(last.replace(/^0x/,''));

    const candidates = [nonce, '0x'+nonce, revHex(nonce), '0x'+revHex(nonce)];
    if (state.extraNonce1 && params.length >= 3) {
      const en2 = String(params[2]||'').replace(/^0x/,'');
      if (en2) {
        candidates.push(padEven(state.extraNonce1)+padEven(en2));
        candidates.push(padEven(state.extraNonce1)+padEven(en2)+padEven(String(params[3]||'')));
      }
    }

    let nodeAccepted = false, nodeResp = null;
    for (const c of [...new Set(candidates)]) {
      try {
        const r = await rpcCall('getwork', [c]);
        nodeResp = r;
        console.log('node reply', (r.raw||'').slice(0,800));
        if (r.parsed && r.parsed.result === true) { nodeAccepted = true; break; }
        if (r.parsed && typeof r.parsed.result === 'object' && r.parsed.result !== null) { nodeAccepted = true; break; }
      } catch (e) { console.warn('rpc err', e && e.message ? e.message : e); }
    }

    if (nodeAccepted) {
      recordShare(worker, true, false);
      sendLine(sock, { id: msg.id || null, result: true, error: null });
      if (nodeResp && nodeResp.parsed && typeof nodeResp.parsed.result === 'object' && nodeResp.parsed.result) {
        const res = nodeResp.parsed.result;
        globalJob = { jobid: res.jobid || '', contenthash: strip0x(res.contenthash || res.contentHash || ''), target: strip0x(res.targetboundary || res.target || ''), extranonce: res.extranonce, extranoncebitsnum: res.extranoncebitsnum, raw: res };
        broadcastJob(globalJob);
      }
      return;
    }

    // node rejected -> local accept for testing
    if (MODE === 'local') {
      recordShare(worker, true, true);
      sendLine(sock, { id: msg.id || null, result: true, error: null });
      console.log('LOCAL accepted share (node rejected)');
      return;
    } else {
      recordShare(worker, false, false);
      sendLine(sock, { id: msg.id || null, result: false, error: 'rejected by node' });
      return;
    }
  }

  // default keepalive
  sendLine(sock, { id: msg.id || null, result: true, error: null });
}

// loops
let pollInterval = null, statsInterval = null;
async function startLoops() {
  const j = await fetchJob();
  if (j) { globalJob = j; broadcastJob(j); }
  pollInterval = setInterval(async () => {
    try {
      const nj = await fetchJob();
      if (nj && (!globalJob || nj.jobid !== globalJob.jobid)) { globalJob = nj; console.log('New job', nj.jobid); broadcastJob(nj); }
    } catch (e) { console.warn('poll err', e && e.message ? e.message : e); }
  }, JOB_POLL_INTERVAL);
  statsInterval = setInterval(()=>printStats(), STATS_PRINT_INTERVAL);
}

function shutdown() {
  console.log('Shutting down proxy...');
  if (pollInterval) clearInterval(pollInterval);
  if (statsInterval) clearInterval(statsInterval);
  try { server.close(); } catch (e) {}
  for (const s of clients) { try { s.end(); } catch (e) {} }
  setTimeout(()=>process.exit(0), 300);
}
process.on('SIGINT', shutdown);
process.on('SIGTERM', shutdown);

server.listen(STRATUM_PORT, STRATUM_HOST, () => {
  console.log(`Listening ${STRATUM_HOST}:${STRATUM_PORT} MODE=${MODE}`);
  startLoops().catch(e => console.warn(e && e.message ? e.message : e));
});