Skip to content

Fix for high concurrency issue #4

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
merged 4 commits into from
Apr 25, 2023
Merged

Fix for high concurrency issue #4

Show file tree
Hide file tree
Changes from 1 commit
Commits
Show all changes
4 commits
Select commit Hold shift + click to select a range
1ed2925
JSONB, PG library, and check max update
billwashere Mar 10, 2018
bde4c6e
formating and review changes
billwashere Mar 17, 2018
698bb62
typo
billwashere Mar 17, 2018
19210ee
Merge branch 'master' into release
billwashere Jun 10, 2018
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
47 changes: 44 additions & 3 deletions README.md
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -8,15 +8,56 @@ Doesn't support queries (yet?).
Moderately experimental. (This drives [Synaptograph](https://www.synaptograph.com)'s backend, and [@nornagon](https://github.com/nornagon) hasn't noticed any issues so far.)


## Requirements

Due to the fix to resolve [high concurency issues](https://github.com/share/sharedb-postgres/issues/1) Postgres 9.5+ is now required.

## Migrating older versions

Older versions of this adaptor used the data type json. You will need to alter the data type prior to using if you are upgrading.

```PLpgSQL
ALTER TABLE ops
ALTER COLUMN operation
SET DATA TYPE jsonb
USING operation::jsonb;

ALTER TABLE snapshots
ALTER COLUMN data
SET DATA TYPE jsonb
USING data::jsonb;
```

## Usage

`sharedb-postgres` wraps native [node-postgres](https://github.com/brianc/node-postgres), and it supports the same configuration options.
`sharedb-postgres-jsonb` wraps native [node-postgres](https://github.com/brianc/node-postgres), and it supports the same configuration options.
Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

can change this back :)


To instantiate a sharedb-postgres wrapper, invoke the module and pass in your
PostgreSQL configuration as an argument. For example:
PostgreSQL configuration as an argument or use environmental arguments.

For example using environmental arugments:

```js
var db = require('sharedb-postgres')();
var backend = require('sharedb')({db: db})
```

Then executing via the command line

```
PGUSER=dbuser PGPASSWORD=secretpassword PGHOST=database.server.com PGDATABASE=mydb PGPORT=5433 npm start
```

Example using an object

```js
var db = require('sharedb-postgres')('postgres://localhost/mydb');
var db = require('sharedb-postgres')({
user: 'dbuser',
host: 'database.server.com',
database: 'mydb',
password: 'secretpassword',
port: 5433,
});
var backend = require('sharedb')({db: db})
```

Expand Down
124 changes: 41 additions & 83 deletions index.js
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -10,6 +10,7 @@ function PostgresDB(options) {
this.closed = false;

this.pg_config = options;
this.pool = new pg.Pool(this.pg_config)
Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Looks like the indentation is off here, please use 2-space indentation :)

};
module.exports = PostgresDB;

Expand All @@ -20,11 +21,6 @@ PostgresDB.prototype.close = function(callback) {
if (callback) callback();
};

function rollback(client, done) {
client.query('ROLLBACK', function(err) {
return done(err);
})
}

// Persists an op and snapshot if it is for the next version. Calls back with
// callback(err, succeeded)
Expand All @@ -39,91 +35,53 @@ PostgresDB.prototype.commit = function(collection, id, op, snapshot, options, ca
* }
* snapshot: PostgresSnapshot
*/
pg.connect(this.pg_config, function(err, client, done) {
if (err) {
done(client);
callback(err);
return;
}
function commit() {
client.query('COMMIT', function(err) {
done(err);
if (err) {
callback(err);
} else {
callback(null, true);
}
})
this.pool.connect((err, client, done) => {
if (err) {
done(client);
callback(err);
return;
}
/*const*/ var query = {
Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I think it's fine to use const directly here, especially since you're using string interpolation 2 lines below--this is ES6

name: 'sdb-commit-op-and-snap',
text: `With snaps as (
Insert into snapshots (collection,doc_id,doc_type, version,data)
Select n.* From ( select $1 c, $2 d, $4 t, $3::integer v, $5::jsonb daa)
n
where v = (select version+1 v from snapshots where collection = $1 and doc_id = $2 for update) or not exists (select 1 from snapshots where collection = $1 and doc_id = $2 for update)
On conflict(collection, doc_id) do update set version = $3, data = $5 , doc_type = $4
Returning version
)
Insert into ops (collection,doc_id, version,operation)
Select n.* From ( select $1 c, $2 t, $3::integer v, $6::jsonb daa)
n
where (v = (select max(version)+1 v from ops where collection = $1 and doc_id = $2) or not exists (select 1 from ops where collection = $1 and doc_id = $2 for update)) and exists (select 1 from snaps)
Returning version`,
Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I found this query really hard to follow, even after reformatting and reindenting it. It's doing a lot of stuff, but I don't know if it's correct--could you explain (perhaps in a comment in the file) what this is doing and why?

Here's the reformatted version I made:

WITH snaps AS (
  INSERT INTO snapshots (collection, doc_id, doc_type, version, data)
  SELECT n.*
  FROM (SELECT $1 c, $2 d, $4 t, $3::integer v, $5::jsonb daa) n 
  WHERE v = (
    SELECT version+1 v
    FROM snapshots
    WHERE collection = $1 AND doc_id = $2
    FOR UPDATE
  ) OR NOT EXISTS (
    SELECT 1
    FROM snapshots
    WHERE collection = $1 AND doc_id = $2
    FOR UPDATE
  )
  ON CONFLICT (collection, doc_id) DO UPDATE SET version = $3, data = $5, doc_type = $4
  RETURNING version
)
INSERT INTO ops (collection, doc_id, version, operation)
SELECT n.* FROM (SELECT $1 c, $2 t, $3::integer v, $6::jsonb daa) n 
WHERE (
  v = (
    SELECT max(version)+1 v
    FROM ops
    WHERE collection = $1 AND doc_id = $2
  ) OR NOT EXISTS (
    SELECT 1
    FROM ops
    WHERE collection = $1 AND doc_id = $2
    FOR UPDATE
  )
) AND EXISTS (SELECT 1 FROM snaps)
RETURNING version

Copy link
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Yeah it non-trivial sql. I've now added the following comment

This query uses common table expression to upsert the snapshot table
(iff the new version is exactly 1 more than the latest table or if
the document id does not exists)

It will then Insert into the ops table if it is exactly 1 more than the
latest table or it the first operation and iff the previous insert into
the snapshot table is successful.

This result of this query the version of the newly inserted operation

If either the ops or the snapshot insert fails then 0 rows are returned

If 0 zeros are return then the callback must return false

values: [collection,id,snapshot.v, snapshot.type, snapshot.data,op]
}
client.query(
'SELECT max(version) AS max_version FROM ops WHERE collection = $1 AND doc_id = $2',
[collection, id],
function(err, res) {
var max_version = res.rows[0].max_version;
if (max_version == null)
max_version = 0;
if (snapshot.v !== max_version + 1) {
return callback(null, false);
}
client.query('BEGIN', function(err) {
client.query(
'INSERT INTO ops (collection, doc_id, version, operation) VALUES ($1, $2, $3, $4)',
[collection, id, snapshot.v, op],
function(err, res) {
if (err) {
// TODO: if err is "constraint violation", callback(null, false) instead
rollback(client, done);
callback(err);
return;
}
if (snapshot.v === 1) {
client.query(
'INSERT INTO snapshots (collection, doc_id, doc_type, version, data) VALUES ($1, $2, $3, $4, $5)',
[collection, id, snapshot.type, snapshot.v, snapshot.data],
function(err, res) {
// TODO:
// if the insert was successful and did insert, callback(null, true)
// if the insert was successful and did not insert, callback(null, false)
// if there was an error, rollback and callback(error)
if (err) {
rollback(client, done);
callback(err);
return;
}
commit();
}
)
} else {
client.query(
'UPDATE snapshots SET doc_type = $3, version = $4, data = $5 WHERE collection = $1 AND doc_id = $2 AND version = ($4 - 1)',
[collection, id, snapshot.type, snapshot.v, snapshot.data],
function(err, res) {
// TODO:
// if any rows were updated, success
// if 0 rows were updated, rollback and not success
// if error, rollback and not success
if (err) {
rollback(client, done);
callback(err);
return;
}
commit();
}
)
}
}
)
})
client.query(query, (err, res) => {
if (err) {
console.log(err.stack)
Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

don't log anything from this library—that's up to the client.

callback(err)
} else if(res.rows.length === 0) {
done(client);
console.log("Unable to commit, not the latest version")
callback(null,false)
}
else {
done(client);
console.log(res.rows[0])
callback(null,true)
}
)
})
})

})
};

// Get the named document from the database. The callback is called with (err,
// snapshot). A snapshot with a version of zero is returned if the docuemnt
// has never been created in the database.
PostgresDB.prototype.getSnapshot = function(collection, id, fields, options, callback) {
pg.connect(this.pg_config, function(err, client, done) {
this.pool.connect(function(err, client, done) {
if (err) {
done(client);
callback(err);
Expand Down Expand Up @@ -173,7 +131,7 @@ PostgresDB.prototype.getSnapshot = function(collection, id, fields, options, cal
//
// Callback should be called as callback(error, [list of ops]);
PostgresDB.prototype.getOps = function(collection, id, from, to, options, callback) {
pg.connect(this.pg_config, function(err, client, done) {
this.pool.connect(function(err, client, done) {
if (err) {
done(client);
callback(err);
Expand Down
2 changes: 1 addition & 1 deletion package.json
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -11,7 +11,7 @@
"keywords": ["sharedb", "sharejs", "share", "postgres"],
"repository": "share/sharedb-postgres",
"dependencies": {
"pg": "^4.5.1",
"pg": "^7.4.1",
"sharedb": "^1.0.0-beta.7"
}
}
10 changes: 10 additions & 0 deletions structure.sql
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -14,3 +14,13 @@ CREATE TABLE snapshots (
data json not null,
PRIMARY KEY (collection, doc_id)
);

ALTER TABLE ops
Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Just fold the data type back into the CREATE TABLE statements

ALTER COLUMN operation
SET DATA TYPE jsonb
USING operation::jsonb;

ALTER TABLE snapshots
ALTER COLUMN data
SET DATA TYPE jsonb
USING data::jsonb;