diff --git a/README.md b/README.md index 60b8bab..a4b5736 100644 --- a/README.md +++ b/README.md @@ -105,6 +105,38 @@ Optionally, you can add a backup process to the DB. Check the ./backup folder. You can build the suggested image via `docker build -t katpool-backup:0.4 .` and uncomment its part in the docker-compose.yml file. We recommend to transfer the database dump files to other location as additional protection. +## Service Account Creation and Credentials for Google Cloud Backup + +### Creating project in google cloud console + - Head over and Login to https://console.cloud.google.com/ + - Go to Topbar right beside the Google Cloud logo + - Create New Project + - Select your newly created project + +### Enabling drive api serivce + - From the navigation menu, select API & services (https://console.cloud.google.com/apis/dashboard) + - Click on ENABLE APIS AND SERVICES (https://console.cloud.google.com/apis/library) + - Go to the Google Workspace in sidebar + - Then click on Google Drive API (https://console.cloud.google.com/apis/library/drive.googleapis.com) + - Click on Enable button + +### Creating the google cloud service account + - Go to (https://console.cloud.google.com/iam-admin/serviceaccounts) + - Click on CREATE SERVICE ACCOUNT, give the service account name, skip the optional fields + +### Creating credentials for the service account + - Go to your newly created service account + - Go to KEYS tab and click on ADD KEY -> Create new key -> Key type : JSON + - Your credentials json file will be downloaded + +### Running cloud backup script + - Add that json file to backup folder as "google-credentials.json" + - Configure the email address to access the dump file in config as "backupEmailAddress" Then execute the below commads: +```bash + cd backup/ + bun run cloudBackup.ts fileName.sql +``` + ## How to install locally using bun (not recommended) To install dependencies: diff --git a/backup/cloudBackup.ts b/backup/cloudBackup.ts new file mode 100644 index 0000000..3c6f16c --- /dev/null +++ b/backup/cloudBackup.ts @@ -0,0 +1,50 @@ +import { createReadStream, access, constants } from 'fs'; +import path from 'path'; +import { google } from 'googleapis'; +import config from '../config/config.json' +import googleCredentials from './google-credentials.json'; + +const SCOPES = ["https://www.googleapis.com/auth/drive.file"]; +const fileNameArgs = process.argv.slice(2); + +async function authorize() { + const jwtClient = new google.auth.JWT( + googleCredentials.client_email, + undefined, + googleCredentials.private_key, + SCOPES + ); + await jwtClient.authorize(); + return jwtClient; +} + +async function uploadFile(authClient: any) { + const drive = google.drive({ version: "v3", auth: authClient }); + + for(let i = 0; i < fileNameArgs.length; i++) { + access(fileNameArgs[i], constants.F_OK, async (err) => { + if (err) { + console.log(`The file ${fileNameArgs[i]} does not exist in the current directory.`); + } else { + const file = await drive.files.create({ + media: { + body: createReadStream(fileNameArgs[i]), + }, + fields: "id", + requestBody: { + name: path.basename(fileNameArgs[i]), + }, + }); + console.log("File Uploaded :", file.data.id); + const backupEmailAddress = config.backupEmailAddress + await drive.permissions.create({ fileId: file.data.id!, requestBody: { type: 'user', role: 'writer', emailAddress: backupEmailAddress } }) + } + }); + } +} + +(async function main() { + const authClient = await authorize(); + await uploadFile(authClient); +})() + diff --git a/config/config.json b/config/config.json index 7c90495..3135b3b 100644 --- a/config/config.json +++ b/config/config.json @@ -13,19 +13,14 @@ "templates": { "cacheSize": 30000 }, - "difficulty": 64, + "difficulty": 4096, "sharesPerMinute": 20, "extraNonceSize": 2, - "varDiff": { - "varDiffStats": true, - "clampPow2": true, - "minDifficulty": 0.5, - "intervalMs": 120000, - "minElapsedSeconds": 30, - "adjustmentFactor": 1.1 - } + "clampPow2": true, + "varDiff": true }, "treasury": { "fee": 5 - } + }, + "backupEmailAddress": "" } \ No newline at end of file diff --git a/go/main.go b/go/main.go index c1e9d56..b82dbfc 100644 --- a/go/main.go +++ b/go/main.go @@ -51,7 +51,7 @@ func NewKaspaAPI(address string, blockWaitTime time.Duration) (*KaspaApi, error) func fetchKaspaAccountFromPrivateKey(network, privateKeyHex string) (string, error) { prefix := util.Bech32PrefixKaspa - if network == "testnet-10" { + if network == "testnet-10" || network == "testnet-11"{ prefix = util.Bech32PrefixKaspaTest } @@ -190,32 +190,32 @@ func main() { templateMutex.Lock() if currentTemplate != nil { - fmt.Printf(` -HashMerkleRoot : %v -AcceptedIDMerkleRoot : %v -UTXOCommitment : %v -Timestamp : %v -Bits : %v -Nonce : %v -DAAScore : %v -BlueWork : %v -BlueScore : %v -PruningPoint : %v -Transactions Length : %v ---------------------------------------- -`, - currentTemplate.Block.Header.HashMerkleRoot, - currentTemplate.Block.Header.AcceptedIDMerkleRoot, - currentTemplate.Block.Header.UTXOCommitment, - currentTemplate.Block.Header.Timestamp, - currentTemplate.Block.Header.Bits, - currentTemplate.Block.Header.Nonce, - currentTemplate.Block.Header.DAAScore, - currentTemplate.Block.Header.BlueWork, - currentTemplate.Block.Header.BlueScore, - currentTemplate.Block.Header.PruningPoint, - len(currentTemplate.Block.Transactions), - ) +// fmt.Printf(` +// HashMerkleRoot : %v +// AcceptedIDMerkleRoot : %v +// UTXOCommitment : %v +// Timestamp : %v +// Bits : %v +// Nonce : %v +// DAAScore : %v +// BlueWork : %v +// BlueScore : %v +// PruningPoint : %v +// Transactions Length : %v +// --------------------------------------- +// `, +// currentTemplate.Block.Header.HashMerkleRoot, +// currentTemplate.Block.Header.AcceptedIDMerkleRoot, +// currentTemplate.Block.Header.UTXOCommitment, +// currentTemplate.Block.Header.Timestamp, +// currentTemplate.Block.Header.Bits, +// currentTemplate.Block.Header.Nonce, +// currentTemplate.Block.Header.DAAScore, +// currentTemplate.Block.Header.BlueWork, +// currentTemplate.Block.Header.BlueScore, +// currentTemplate.Block.Header.PruningPoint, +// len(currentTemplate.Block.Transactions), +// ) } else { fmt.Println("No block template fetched yet.") } diff --git a/package.json b/package.json index 6940a2d..d1bdc9d 100644 --- a/package.json +++ b/package.json @@ -21,6 +21,7 @@ "denque": "^2.1.0", "dotenv": "^16.4.5", "fs": "^0.0.1-security", + "googleapis": "^144.0.0", "json-bigint": "^1.0.0", "lmdb": "^3.0.11", "node-cron": "^3.0.3", diff --git a/src/stratum/index.ts b/src/stratum/index.ts index 9010e50..54a7131 100644 --- a/src/stratum/index.ts +++ b/src/stratum/index.ts @@ -41,9 +41,10 @@ export default class Stratum extends EventEmitter { this.monitoring.log(`Stratum: Initialized with difficulty ${this.difficulty}`); // Start the VarDiff thread - const varDiffStats = config.stratum.varDiff.varDiffStats || true; // Enable logging of VarDiff stats - const clampPow2 = config.stratum.varDiff.clampPow2 || true; // Enable clamping difficulty to powers of 2 - this.sharesManager.startVardiffThread(sharesPerMin, varDiffStats, clampPow2); + const clampPow2 = config.stratum.clampPow2 || true; // Enable clamping difficulty to powers of 2 + const varDiff = config.stratum.varDiff || false; // Enable variable difficulty + if (varDiff) + this.sharesManager.startVardiffThread(sharesPerMin, clampPow2); this.extraNonceSize = Math.min(Number(config.stratum.extraNonceSize), 3 ) || 0; this.maxExtranonce = Math.pow(2, 8 * Math.min(this.extraNonceSize, 3)) - 1; @@ -68,17 +69,23 @@ export default class Stratum extends EventEmitter { this.subscriptors.forEach((socket) => { if (socket.readyState === "closed") { this.subscriptors.delete(socket); - } else { - this.reflectDifficulty(socket); + } else { + socket.data.workers.forEach((worker, _) => { + var varDiff = this.sharesManager.getClientVardiff(worker) + if (varDiff != socket.data.difficulty && varDiff != 0) { + this.monitoring.log(`Stratum: Updating VarDiff for ${worker.name} from ${socket.data.difficulty} to ${varDiff}`); + this.sharesManager.updateSocketDifficulty(worker.address, varDiff) + this.reflectDifficulty(socket) + this.sharesManager.startClientVardiff(worker) + } + }); + socket.write(tasksData[socket.data.encoding] + '\n'); } }); } reflectDifficulty(socket: Socket) { - if (socket.data.encoding === Encoding.Bitmain) { - socket.data.difficulty = 4096 - } const event: Event<'mining.set_difficulty'> = { method: 'mining.set_difficulty', params: [socket.data.difficulty] @@ -197,7 +204,7 @@ export default class Stratum extends EventEmitter { const minerData = this.sharesManager.getMiners().get(worker.address); const workerDiff = minerData?.workerStats.minDiff; const socketDiff = socket.data.difficulty; - if (DEBUG) this.monitoring.debug(`Stratum: Current difficulties - Worker: ${workerDiff}, Socket: ${socketDiff}`); + if (DEBUG) this.monitoring.debug(`Stratum: Current difficulties , Worker Name: ${minerId} - Worker: ${workerDiff}, Socket: ${socketDiff}`); const currentDifficulty = workerDiff || socketDiff; if (DEBUG) this.monitoring.debug(`Stratum: Adding Share - Address: ${address}, Worker Name: ${name}, Hash: ${hash}, Difficulty: ${currentDifficulty}`); // Add extranonce to noncestr if enabled and submitted nonce is shorter than diff --git a/src/stratum/sharesManager.ts b/src/stratum/sharesManager.ts index 0b173ca..d30bf41 100644 --- a/src/stratum/sharesManager.ts +++ b/src/stratum/sharesManager.ts @@ -1,6 +1,7 @@ import type { Socket } from 'bun'; import { calculateTarget } from "../../wasm/kaspa"; import { Pushgateway } from 'prom-client'; +import { type Worker } from './server'; import type { RegistryContentType } from 'prom-client'; import { stringifyHashrate, getAverageHashrateGHs } from './utils'; import Monitoring from '../monitoring'; @@ -42,6 +43,9 @@ type MinerData = { workerStats: WorkerStats }; +const varDiffThreadSleep: number = 10 +const zeroDateMillS: number = new Date(0).getMilliseconds() + type Contribution = { address: string; difficulty: number; @@ -140,20 +144,6 @@ export class SharesManager { } }; this.miners.set(address, minerData); - } else { - // // Atomically update worker stats - // minerData.workerStats.sharesFound++; - // minerData.workerStats.varDiffSharesFound++; - // minerData.workerStats.lastShare = timestamp; - // minerData.workerStats.minDiff = currentDifficulty; - - // // Update recentShares with the new share - // minerData.workerStats.recentShares.push({ timestamp: Date.now(), difficulty: currentDifficulty }); - - // const windowSize = 10 * 60 * 1000; // 10 minutes window - // while (minerData.workerStats.recentShares.length > 0 && Date.now() - minerData.workerStats.recentShares.peekFront()!.timestamp > windowSize) { - // minerData.workerStats.recentShares.shift(); - // } } const state = templates.getPoW(hash); @@ -190,9 +180,6 @@ export class SharesManager { minerData.workerStats.varDiffSharesFound++; minerData.workerStats.lastShare = timestamp; minerData.workerStats.minDiff = currentDifficulty; - if (encoding === Encoding.Bitmain) { - minerData.workerStats.minDiff = 4096 - } // Update recentShares with the new share minerData.workerStats.recentShares.push({ timestamp: Date.now(), difficulty: currentDifficulty }); @@ -201,40 +188,6 @@ export class SharesManager { while (minerData.workerStats.recentShares.length > 0 && Date.now() - minerData.workerStats.recentShares.peekFront()!.timestamp > windowSize) { minerData.workerStats.recentShares.shift(); } - // Implement variable difficulty - this.updateDifficulty(minerId); - } - - private updateDifficulty(minerId: string): void { - const workerStats = this.miners.get(minerId)?.workerStats; - if (!workerStats) return; - - const now = Date.now(); - const elapsedMs = now - workerStats.varDiffStartTime; - - if (elapsedMs >= 120000) { // 120000ms = 2 minutes - const shareRate = workerStats.varDiffSharesFound / (elapsedMs / 1000); - const targetShareRate = 60 / 60; // 60 shares per minute - - let newDifficulty = workerStats.minDiff; - - if (shareRate > targetShareRate * 1.1) { - newDifficulty *= 1.1; - } else if (shareRate < targetShareRate * 0.9) { - newDifficulty /= 1.1; - } - - newDifficulty = Math.max(newDifficulty, 1); - - if (newDifficulty !== workerStats.minDiff) { - workerStats.minDiff = newDifficulty; - this.monitoring.log(`SharesManager: Updated difficulty for ${minerId} to ${newDifficulty}`); - varDiff.labels(minerId).set(newDifficulty); - } - - workerStats.varDiffStartTime = now; - workerStats.varDiffSharesFound = 0; - } } startStatsThread() { @@ -351,58 +304,6 @@ export class SharesManager { this.contributions.clear(); } - startVardiffThread(sharesPerMin: number, varDiffStats: boolean, clampPow2: boolean) { - const intervalMs = 120000; // Run every 2 minutes - const minElapsedSeconds = 30; // Minimum 30 seconds between adjustments - const adjustmentFactor = 1.1; // 10% adjustment - const minDifficulty = 1; // Minimum difficulty - - setInterval(() => { - const now = Date.now(); - - this.miners.forEach((minerData, address) => { - const stats = minerData.workerStats; - const elapsedSeconds = (now - stats.varDiffStartTime) / 1000; - if (elapsedSeconds < minElapsedSeconds) return; - - const sharesFound = stats.varDiffSharesFound; - const shareRate = (sharesFound / elapsedSeconds) * 60; // Convert to per minute - const targetRate = sharesPerMin; - - if (DEBUG) this.monitoring.debug(`SharesManager - VarDiff for ${stats.workerName}: sharesFound: ${sharesFound}, elapsedSeconds: ${elapsedSeconds}, shareRate: ${shareRate}, targetRate: ${targetRate}, currentDiff: ${stats.minDiff}`); - - let newDiff = stats.minDiff; - - if (shareRate > targetRate * 1.2) { - newDiff = stats.minDiff * adjustmentFactor; - } else if (shareRate < targetRate * 0.8) { - newDiff = stats.minDiff / adjustmentFactor; - } - - if (clampPow2) { - newDiff = Math.pow(2, Math.round(Math.log2(newDiff))); - } - - newDiff = Math.max(newDiff, minDifficulty); - - if (newDiff !== stats.minDiff) { - this.monitoring.debug(`SharesManager: VarDiff - Adjusting difficulty for ${stats.workerName} from ${stats.minDiff} to ${newDiff}`); - stats.minDiff = newDiff; - this.updateSocketDifficulty(address, newDiff); - } else { - this.monitoring.debug(`SharesManager: VarDiff - No change in difficulty for ${stats.workerName} (current difficulty: ${stats.minDiff})`); - } - - stats.varDiffSharesFound = 0; - stats.varDiffStartTime = now; - - if (varDiffStats) { - this.monitoring.log(`SharesManager: VarDiff for ${stats.workerName}: sharesFound: ${sharesFound}, elapsed: ${elapsedSeconds.toFixed(2)}, shareRate: ${shareRate.toFixed(2)}, newDiff: ${stats.minDiff}`); - } - }); - }, intervalMs); - } - updateSocketDifficulty(address: string, newDifficulty: number) { const minerData = this.miners.get(address); if (minerData) { @@ -422,4 +323,148 @@ export class SharesManager { this.lastAllocationTime = currentTime; return shares; } + + async sleep(ms: number) { + return new Promise(resolve => setTimeout(resolve, ms)); + } + + startVardiffThread(expectedShareRate: number, clamp: boolean): void { + // 20 shares/min allows a ~99% confidence assumption of: + // < 100% variation after 1m + // < 50% variation after 3m + // < 25% variation after 10m + // < 15% variation after 30m + // < 10% variation after 1h + // < 5% variation after 4h + var windows: number[] = [1, 3, 10, 30, 60, 240, 0] + var tolerances: number[] = [1, 0.5, 0.25, 0.15, 0.1, 0.05, 0.05] + + setInterval(async () => { + await this.sleep(varDiffThreadSleep * 1000); + + // don't like locking entire stats struct - risk should be negligible + // if mutex is ultimately needed, should move to one per client + // sh.statsLock.Lock() + + var stats: string = "\n=== vardiff ===================================================================\n\n" + stats += " worker name | diff | window | elapsed | shares | rate \n" + stats += "-------------------------------------------------------------------------------\n" + + var statsLines: string[] = [] + var toleranceErrs: string[] = [] + + for (const [address, minerData] of this.miners) { + const workerStats = minerData.workerStats; + var worker: string = workerStats.workerName + if (workerStats.varDiffStartTime == zeroDateMillS) { + // no vardiff sent to client + toleranceErrs = toleranceErrs.concat(toleranceErrs, `no diff sent to client ${worker}`) + continue + } + + var diff : number = workerStats.minDiff + var shares: number = workerStats.varDiffSharesFound + var duration: number = (Date.now() - workerStats.varDiffStartTime) / 60000 + var shareRate: number = shares / duration + var shareRateRatio: number = shareRate / expectedShareRate + var window: number = windows[workerStats.varDiffWindow] + var tolerance: number = tolerances[workerStats.varDiffWindow] + + statsLines = statsLines.concat(` ${worker.padEnd(14)}| ${diff.toFixed(2).padStart(11)} | ${window.toString().padStart(8)} | ${duration.toFixed(2).padStart(10)} | ${shares.toString().padStart(11)} | ${shareRate.toFixed(2).padStart(9)}\n`) + + // check final stage first, as this is where majority of time spent + if (window == 0) { + if (Math.abs(1 - shareRateRatio) >= tolerance) { + // final stage submission rate OOB + toleranceErrs = toleranceErrs.concat(toleranceErrs, `${worker} final share rate ${shareRate} exceeded tolerance (+/- ${tolerance*100}%)`) + this.updateVarDiff(workerStats, diff * shareRateRatio, clamp) + } + continue + } + + // check all previously cleared windows + var i: number = 1 + for (; i < workerStats.varDiffWindow; ) { + if (Math.abs(1 - shareRateRatio) >= tolerances[i]) { + // breached tolerance of previously cleared window + toleranceErrs = toleranceErrs.concat(toleranceErrs, `${worker} share rate ${shareRate} exceeded tolerance (+/- ${tolerances[i]*100}%) for ${windows[i]}m window`) + this.updateVarDiff(workerStats, diff * shareRateRatio, clamp) + break + } + i++ + } + if (i < workerStats.varDiffWindow) { + // should only happen if we broke previous loop + continue + } + + // check for current window max exception + if (shares >= window * expectedShareRate * (1 + tolerance)) { + // submission rate > window max + toleranceErrs = toleranceErrs.concat(toleranceErrs, `${worker} share rate ${shareRate} exceeded upper tolerance (+/- ${tolerances[i]*100}%) for ${windows[i]}m window`) + this.updateVarDiff(workerStats, diff*shareRateRatio, clamp) + continue + } + + // check whether we've exceeded window length + if (duration >= window) { + // check for current window min exception + if (shares <= window * expectedShareRate * (1 - tolerance)) { + // submission rate < window min + toleranceErrs = toleranceErrs.concat(toleranceErrs, `${worker} share rate ${shareRate} exceeded lower tolerance (+/- ${tolerances[i]*100}%) for ${windows[i]}m window`) + this.updateVarDiff(workerStats, diff * Math.max(shareRateRatio, 0.1), clamp) + continue + } + + workerStats.varDiffWindow++ + } + } + + statsLines.sort() + stats += statsLines + "\n" + stats += `\n\n===============================================================================\n` + stats += `\n${toleranceErrs}\n\n\n` + if (DEBUG) { + this.monitoring.debug(stats) + } + + // sh.statsLock.Unlock() + }, varDiffThreadSleep * 1000); + } + + // (re)start vardiff tracker + startVarDiff(stats: WorkerStats) { + if (stats.varDiffStartTime == zeroDateMillS) { + stats.varDiffSharesFound = 0 + stats.varDiffStartTime = Date.now() + } + } + + // update vardiff with new mindiff, reset counters, and disable tracker until + // client handler restarts it while sending diff on next block + updateVarDiff(stats : WorkerStats, minDiff: number, clamp: boolean): number { + if (clamp) { + minDiff = Math.pow(2, Math.floor(Math.log2(minDiff))) + } + + var previousMinDiff = stats.minDiff + var newMinDiff = Math.max(0.125, minDiff) + if (newMinDiff != previousMinDiff) { + this.monitoring.log(`updating vardiff to ${newMinDiff} for client ${stats.workerName}`) + stats.varDiffStartTime = zeroDateMillS + stats.varDiffWindow = 0 + stats.minDiff = newMinDiff + } + return previousMinDiff + } + + startClientVardiff(worker: Worker) { + const stats = this.getOrCreateWorkerStats(worker.name, this.miners.get(worker.address)!); + this.startVarDiff(stats) + } + + getClientVardiff(worker: Worker): number { + const stats = this.getOrCreateWorkerStats(worker.name, this.miners.get(worker.address)!); + return stats.minDiff + } } \ No newline at end of file diff --git a/src/stratum/templates/index.ts b/src/stratum/templates/index.ts index 9beeb51..1c921f1 100644 --- a/src/stratum/templates/index.ts +++ b/src/stratum/templates/index.ts @@ -64,7 +64,7 @@ export default class Templates { if (report.report.type == "success") { metrics.updateGaugeInc(paidBlocksGauge, [minerId, this.address]); - if (DEBUG) this.monitoring.debug(`Templates: the block by miner ${minerId} has been accepted`) + if (DEBUG) this.monitoring.debug(`Templates: the block by miner ${minerId} has been accepted with hash : ${newHash}`) } else { // Failed if (DEBUG) this.monitoring.debug(`Templates: the block by ${minerId} has been rejected, reason: ${report.report.reason}`) } @@ -82,7 +82,7 @@ export default class Templates { // })).block as IRawBlock; const templateChannel = config.redis_channel - this.subscriber.subscribe(templateChannel, (message) => { + this.subscriber.subscribe(templateChannel, (message) => { const fetchedTemplate = JSON.parse(message) const blockTemplate = { header: fetchedTemplate.Block.Header,