Блог

Прежде чем я передам что-либо в базу данных, я хочу, чтобы все мои обещания обновления разрешились; в противном случае я выполняю откат. Другими словами, я хочу атомарности. Я полагаю, я мог бы обработать откат, удалив строки, но это сопряжено со своими рисками. Я заметил, что если в каком-либо из обещаний есть ошибка, данные все равно обновляются в базе данных. Что я делаю не так?

Я написал простую программу, чтобы проиллюстрировать проблему. Это основной процесс:

 const db = require('./db.js');
const async = require('async');

let insertList = [];
for (let i = 0; i<3; i  ) {
    insertList.push(i);
}

async function func1 () {
    return new Promise((resolve, reject) => {
        console.log("In Func1");
        async.forEachOf(insertList, function(value, key, callback) {
            console.log('>>>>'   value   '<<<<<<'   key );
            db.insertOne('coll1', {value}).then(() => {
                callback();
            }).catch(err => {callback(err)})
        }, function(err) {
            // if any of the file processing produced an error, err would equal that error
            if( err ) {
                // One of the iterations produced an error.
                // All processing will now stop.
                console.log('err:', err);
                reject(err);
            } else {
                console.log('Col1 All inserts have been processed successfully');
                resolve("Success");
            }
        });
    })
}

function func2 () {
    return new Promise((resolve, reject) => {
        console.log("In Func2");
        async.forEachOf(insertList, function(value, key, callback) {
            console.log('>>>>'   value   '<<<<<<'   key );
            db.insertOne('coll2', {value}).then(() => {
                callback();
            }).catch(err => {callback(err)})
        }, function(err) {
            // if any of the file processing produced an error, err would equal that error
            if( err ) {
                // One of the iterations produced an error.
                // All processing will now stop.
                console.log('err:', err);
                reject(err);
            } else {
                console.log('Col2 All inserts have been processed successfully');
                resolve("Success");
            }
        });
    })
}

function func3 () {
    return new Promise((resolve, reject) => {
        console.log("In Func3");
        async.forEachOf(insertList, function(value, key, callback) {
            console.log('>>>>'   value   '<<<<<<'   key );
            if(key === 1) {
                value = 'a';
            }
            db.insertOne('coll3', {value}).then(() => {
                callback();
            }).catch(err => {callback(err)})
        }, function(err) {
            // if any of the file processing produced an error, err would equal that error
            if( err ) {
                // One of the iterations produced an error.
                // All processing will now stop.
                console.log('err:', err);
                reject(err);
            } else {
                console.log('Col3 All inserts have been processed successfully');
                resolve("Success");
            }
        });
    })
}

db.connect().then((pool) => {
    pool.getConnection((err, connection) =>{
        if (err)
            return console.error(err);
        else {

        }
            connection.beginTransaction((err) => {
                if (err) {
                    return console.error(err);
                }

                let func1Promise = new Promise((resolve, reject) => {func1().then(() => {
                    console.log("Func1 complete");
                    resolve("Func1 complete");
                }).catch((err) => {
                    console.error("Func1 ERROR: ", err);
                    reject("Func1 ERROR: ", err);
                })});

                let func2Promise = new Promise((resolve, reject) => {func2().then(() => {
                    console.log("Func2 complete");
                    resolve("Func2 complete");
                }).catch((err) => {
                    console.error("Func2 ERROR: ", err);
                    reject("Func2 ERROR: ", err);
                })});

                let func3Promise = new Promise((resolve, reject) => {func3().then(() => {
                    console.log("Func3 complete");
                    resolve("Func3 complete");
                }).catch((err) => {
                    console.error("Func3 ERROR: ", err);
                    reject("Func3 ERROR: ", err);
                })});

                Promise.all([func1Promise, func2Promise, func3Promise])
                    .then(()=> {
                        console.log("All Processes completed successfully.");
                        connection.commit(err => {
                            if (err) {
                                connection.rollback(() => {
                                    throw err;
                                });
                            }
                            console.log('Commit Complete.');
                            connection.release();
                        });
                    })
                    .catch((err)=> {
                        console.error(err);
                        console.error("An update process has failed.");
                        connection.rollback(() => {
                            console.error(err);
                            connection.release();
                        });
                    })

            });
    })
});
  

The db.js выглядит так:

 const mysql      = require('mysql');
const config = {
    db: {
        host: 'localhost',
        user: process.env.DBUSER,
        password: process.env.DBPASSWORD,
        database: 'test',
    }
};

var pool;
class DB {

    constructor(host, user, password, database) {
        this.host = host;
        this.user = user;
        this.password = password;
        this.database = database;
    }

    connect() {
        return new Promise((resolve, reject) => {
            pool = mysql.createPool({
                connectionLimit: 10,
                host     : this.host,
                user     : this.user,
                password : this.password,
                database : this.database
            });
            resolve(pool);
        });
    }

    objToArray(obj) {
        let arr = obj instanceof Array;

        return (arr ? obj : Object.keys(obj)).map((i) => {
            let val = arr ? i : obj[i];
            if(typeof val === 'object' amp;amp; val !== null)
                return this.objToArray(val);
            else
                return val;
        });
    }

    insertOne(collection, insertObj) {
        return new Promise((resolve, reject) => {
            pool.getConnection((err, connection) => {
                if (err) {
                    resolve(err);
                } else {
                    let sql = "INSERT INTO "   collection   " VALUES (?)";
                    // Convert the array of objects into an array of arrays.
                    let responseJson = this.objToArray(insertObj);

                    // The query object expects an array of objects so you pass in 'responseJson' as is
                    console.log(responseJson);
                    connection.query(sql, [responseJson], (err, result) => {
                        if (err) {
                            console.error(err);
                            return reject(err);
                        }
                        //console.log(result);
                        resolve("SUCCESS: object inserted into database");
                    });
                }
            });
        });
    }

}

const db = new DB(config.db.host, config.db.user, config.db.password, config.db.database);
Object.freeze(db);

module.exports = db;
  

Моя база данных «test» проста и состоит из 3 таблиц: coll1, coll2, coll3, и в каждой есть поле on, которое имеет тип int. В третьей функции я заменяю 1 на ‘a’, это вызывает ошибку, и код улавливает эту ошибку и пытается выполнить откат, который не работает. Если я установлю точку останова после выполнения func1 и проверю базу данных, значения уже есть в базе данных.

Вот версия MySQL, которую я запускаю:

 Variable_name,Value
innodb_version,8.0.11
protocol_version,10
slave_type_conversions,
tls_version,"TLSv1,TLSv1.1,TLSv1.2"
version,8.0.11
version_comment,"MySQL Community Server - GPL"
version_compile_machine,x86_64
version_compile_os,macos10.13
version_compile_zlib,1.2.11
  

Я использую следующие пакеты NPM в узле:

 "async": "^2.6.2",
"mysql": "^2.15.0"
  

Вы создаете транзакцию для соединения, созданного в вашей тестовой программе, но ваш db.js insertOne получает новое соединение из пула, в котором нет транзакции. Вы должны передавать соединение, которое вы создали в тестовой программе.