MongoExpiredSessionError:不能使用已经结束的会话

Question

我正在尝试从数据库集合中的现有文档创建新文档。我将原始文档中的一个字段扁平化。在使用mongosh时，这个过程可以完美地工作，但是现在我必须在gcp函数上运行这个脚本。所以我需要在脚本中包含所有内容。

我设法连接到db，将扁平文档添加到新的集合，但是当我关闭连接时，它给了我错误:

MongoExpiredSessionError: Cannot use a session that has ended

并且它永远不会将文档添加到新的集合中。

我已经弄清楚，当我从代码中删除DB.close()时，它添加了文档，但程序从未完成/退出。

我已经将每个await文档放在它与DB交互的地方，但它没有多大帮助。这是我的完整代码。main函数调用connectMongo，它连接到mongo数据库并运行execute函数。这个execute函数从原始集合中获取文档，并将它们一个接一个地传递给flatten函数。flatten函数执行平坦化操作，并将新文档添加到新集合中。

代码如下:打印语句在所有情况下(w/&w/outDB.close())工作并打印出每个

const {MongoClient} = require('mongodb');
const uri = "mongodb://<myURI>"

const DB = new MongoClient(uri);
var collections;
//---------------------------------------------------------------------------------------------
execute = async function(db){

docs = await db.collection('observations6').find()

await docs.forEach(flatten)

}
//---------------------------------------------------------------------------------------------
flatten = async function(myDoc){
forms = myDoc.properties.forms
for(var i = 0; i < forms.length; i++){
FormObj = {
parent_id : myDoc._id+'_form_'+i,
eventId: myDoc.eventId,
userId: myDoc.userId,
deviceId: myDoc.deviceId,
createdAt: myDoc.createdAt,
lastModified : myDoc.lastModified,
type: myDoc.type,
geometry: myDoc.geometry,
favoriteUserIds: myDoc.favoriteUserIds,
states: myDoc.states,
attachments: myDoc.attachments,
formId: forms[i]['formId'],
}
console.log('Created form object')
field_count = 0
retry = 0
while (retry <= 10){
if (!forms[i].hasOwnProperty('field'+field_count)){
field_count += 1
}
retry += 1
}
console.log(`Did ${field_count} retry(s)`)
while(forms[i].hasOwnProperty('field'+field_count)){
FormObj['field'+field_count] = forms[i]['field'+field_count]
field_count+=1
}
console.log('Added field properties')

parent_id = myDoc._id + '_form_' + i
await collections.collection('observations6_flatten').updateOne({parent_id},{$set: {FormObj}}, {upsert:true})
console.log('Added object to collection')

}

}
//---------------------------------------------------------------------------------------------
async function connectMongo() {

try {
// Connect to the MongoDB cluster
await DB.connect();

// Make the appropriate DB calls
collections = await DB.db('magedb')
//console.log(collections)
await execute(collections)


} catch (e) {
console.error(e);
} 

}
//---------------------------------------------------------------------------------------------
//Main call
main = async function(){
await connectMongo()
.then(()=>{
DB.close()
})
}
main()

这是它返回的错误

aortiz@WW-593 MINGW64 ~/Documents/GitHub/engineering_etl (main)
$ node flattening/flattening.js
Created form object
Did 11 retry(s)
Added field properties
Created form object
Did 0 retry(s)
Added field properties
Created form object
Did 1 retry(s)
Added field properties
Created form object
Did 0 retry(s)
Added field properties
Created form object
Did 0 retry(s)
Added field properties
Created form object
Did 0 retry(s)
Added field properties
Created form object
Did 0 retry(s)
Added field properties
C:UsersaortizDocumentsGitHubengineering_etlnode_modulesmongodblibsessions.js:655
return new error_1.MongoExpiredSessionError();
^
MongoExpiredSessionError: Cannot use a session that has ended
at applySession (C:UsersaortizDocumentsGitHubengineering_etlnode_modulesmongodblibsessions.js:655:16)        
at Connection.command (C:UsersaortizDocumentsGitHubengineering_etlnode_modulesmongodblibcmapconnection.js:281:53)
at C:UsersaortizDocumentsGitHubengineering_etlnode_modulesmongodblibsdamserver.js:210:18
at Object.callback (C:UsersaortizDocumentsGitHubengineering_etlnode_modulesmongodblibcmapconnection_pool.js:327:13)
at ConnectionPool.processWaitQueue (C:UsersaortizDocumentsGitHubengineering_etlnode_modulesmongodblibcmapconnection_pool.js:506:33)
at C:UsersaortizDocumentsGitHubengineering_etlnode_modulesmongodblibcmapconnection_pool.js:183:37
at processTicksAndRejections (node:internal/process/task_queues:78:11) {
[Symbol(errorLabels)]: Set(0) {}
}

当我移除DB.close()时，它返回

aortiz@WW-593 MINGW64 ~/Documents/GitHub/engineering_etl (main)
$ node flattening/flattening.js
Created form object
Did 11 retry(s)
Added field properties
Created form object
Did 0 retry(s)
Added field properties
Created form object
Did 1 retry(s)
Added field properties
Created form object
Did 0 retry(s)
Added field properties
Created form object
Did 0 retry(s)
Added field properties
Created form object
Did 0 retry(s)
Added field properties
Created form object
Did 0 retry(s)
Added field properties
Added object to collection
Added object to collection
Added object to collection
Added object to collection
Added object to collection
Added object to collection
Added object to collection

和未完成/退出脚本

编辑

现在，当我删除DB.close()时，我可以更好地查看输出，我看到Added Object to collection打印语句被打印出来，当我有DB.close()时，它在此之前停止。因此，这一定意味着错误发生在以下行:

await collections.collection('observations6_flatten').updateOne({parent_id},{$set: {FormObj}}, {upsert:true})

但是连接在此点之前结束到底是怎么回事呢?

Answer 1

正如在评论中提到的，您确实需要破解承诺和async/await语法https://developer.mozilla.org/en-US/docs/Web/JavaScript/Guide/Using_promises。我建议在事件循环的背景下学习它:https://developer.mozilla.org/en-US/docs/Web/JavaScript/EventLoop, https://nodejs.org/en/docs/guides/event-loop-timers-and-nexttick/等。

await docs.forEach(flatten)

不像你想象的那样。

https://developer.mozilla.org/en-US/docs/Web/JavaScript/Reference/Global_Objects/Array/forEach是一个同步呼叫它的工作原理与没有await相同。它迭代docs数组，将flatten函数一次添加到所有项目的队列中，并解析"promise"，即将控制权返回给调用await execute(collections)，而不需要等待内部的任何promise。

await execute(collections)是函数中的最后一条语句，所以一旦这个promise被解决，它将控制权返回给调用者

await connectMongo()
.then(()=>{
DB.close()
})

在任何await collections.collection('observations6_flatten').updateOne有机会控制之前关闭连接。

您需要使用https://developer.mozilla.org/en-US/docs/Web/JavaScript/Reference/Global_Objects/Promise/all:

execute = async function(db){

docs = await db.collection('observations6').find().toArray();

return await Promise.all(docs.map(flatten))    
} 

或显式遍历文档:

execute = async function(db){

docs = await db.collection('observations6').find().toArray();

for(const doc of docs) { 
await flatten(doc)
}    
} 

Promise.all立即启动所有更新，并在flatten解决所有项目时解决承诺。for-await循环要慢得多，因为它一次更新一个文档。

Promise.all的完整工作示例(我简化了无关的平坦化逻辑):https://replit.com/@blex18/74891663mongoexpiredsessionerror-cannot-use-a-session?v=1

控制台输出:

{ _id: new ObjectId("63ada717dc7041cce1c67a86") }
Created form object
Did 0 retry(s)
Added field properties
{ _id: new ObjectId("63ada723dc7041cce1c67a87") }
Created form object
Did 0 retry(s)
Added field properties
{ _id: new ObjectId("63ada725dc7041cce1c67a88") }
Created form object
Did 0 retry(s)
Added field properties
Added object to collection
Created form object
Did 0 retry(s)
Added field properties
Added object to collection
Created form object
Did 0 retry(s)
Added field properties
Added object to collection
Added object to collection
Created form object
Did 0 retry(s)
Added field properties
Added object to collection
Created form object
Did 0 retry(s)
Added field properties
Added object to collection
Created form object
Did 0 retry(s)
Added field properties
Added object to collection
Created form object
Did 0 retry(s)
Added field properties
Added object to collection
Added object to collection

Answer 2

await与。then()不同，如果你想了解更多关于承诺链接的信息，请阅读这里

const {MongoClient} = require('mongodb');
const uri = "mongodb://<myURI>"
const client = new MongoClient(uri);
async function connectMongo() {
try {
await client.connect();
console.log('Connected successfully to server');
// Make the appropriate DB calls
const collections = await client.db('magedb');
console.log('Collections', collections);
await execute(collections);
} catch (e) {
console.error(e);
} 
}
async function execute(myDoc) {
forms = myDoc.properties.forms
for(var i = 0; i < forms.length; i++){
FormObj = {
parent_id : myDoc._id+'_form_'+i,
eventId: myDoc.eventId,
userId: myDoc.userId,
deviceId: myDoc.deviceId,
createdAt: myDoc.createdAt,
lastModified : myDoc.lastModified,
type: myDoc.type,
geometry: myDoc.geometry,
favoriteUserIds: myDoc.favoriteUserIds,
states: myDoc.states,
attachments: myDoc.attachments,
formId: forms[i]['formId'],
}
console.log('Created form object')
field_count = 0
retry = 0
while (retry <= 10){
if (!forms[i].hasOwnProperty('field'+field_count)){
field_count += 1
}
retry += 1
}
console.log(`Did ${field_count} retry(s)`)
while(forms[i].hasOwnProperty('field'+field_count)){
FormObj['field'+field_count] = forms[i]['field'+field_count]
field_count+=1
}
console.log('Added field properties')

parent_id = myDoc._id + '_form_' + i;
try {
await client.db().collection('observations6_flatten').updateOne({parent_id},{$set: {FormObj}}, {upsert:true});
console.log('Added object to collection');
} catch (error) {
throw error; // still want to crash
}
}

}
connectMongo()
.then(console.log)
.catch(console.error)
.finally(() => client.close());