live, common: Add support for downloading live databases

Author: justinclift

api/handlers.go

@@ -206,9 +206,9 @@ func checkAuth(w http.ResponseWriter, r *http.Request) (loggedInUser string, err
 }
 
 // collectInfo is an internal function which:
-//   1. Authenticates incoming requests
-//   2. Extracts the database owner, name, and commit ID from the request
-//   3. Checks permissions
+//  1. Authenticates incoming requests
+//  2. Extracts the database owner, name, and commit ID from the request
+//  3. Checks permissions
 func collectInfo(w http.ResponseWriter, r *http.Request) (loggedInUser, dbOwner, dbName, commitID string, httpStatus int, err error) {
 	// Authenticate the request
 	loggedInUser, err = checkAuth(w, r)
@@ -241,9 +241,10 @@ func collectInfo(w http.ResponseWriter, r *http.Request) (loggedInUser, dbOwner,
 }
 
 // collectInfoAndOpen is an internal function which:
-//   1. Calls collectInfo() to authenticate the request + collect the user/database/commit/etc details
-//   2. Fetches the database from Minio
-//   3. Opens the database, returning the connection handle
+//  1. Calls collectInfo() to authenticate the request + collect the user/database/commit/etc details
+//  2. Fetches the database from Minio
+//  3. Opens the database, returning the connection handle
+//
 // This function exists purely because this code is common to most of the handlers
 func collectInfoAndOpen(w http.ResponseWriter, r *http.Request) (sdb *sqlite.Conn, httpStatus int, err error) {
 	// Authenticate the request and collect details for the requested database

api/main.go

@@ -297,10 +297,10 @@ func MQCreateResponse(msg amqp.Delivery, channel *amqp.Channel, nodeName, result
 	return
 }
 
-// MQErrorResponse sends an error message in response to an AMQP request
+// MQErrorResponse sends an error message in response to an AMQP request.
 // It is probably only useful for returning errors that occur before we've decoded the incoming AMQP
 // request to know what type it is
-func MQErrorResponse(msg amqp.Delivery, channel *amqp.Channel, nodeName string, errMsg string) (err error) {
+func MQErrorResponse(requestType string, msg amqp.Delivery, channel *amqp.Channel, nodeName string, errMsg string) (err error) {
 	// Construct the response
 	resp := LiveDBErrorResponse{
 		Node:  nodeName,
@@ -327,7 +327,7 @@ func MQErrorResponse(msg amqp.Delivery, channel *amqp.Channel, nodeName string,
 	}
 	msg.Ack(false)
 	if AmqpDebug {
-		log.Printf("[NOT-YET-DETERMINED] Live node '%s' responded with ACK to message with correlationID: '%s', msg.ReplyTo: '%s'", nodeName, msg.CorrelationId, msg.ReplyTo)
+		log.Printf("[%s] Live node '%s' responded with ACK to message with correlationID: '%s', msg.ReplyTo: '%s'", requestType, nodeName, msg.CorrelationId, msg.ReplyTo)
 	}
 	return
 }
@@ -467,6 +467,7 @@ func MQQueryResponse(msg amqp.Delivery, channel *amqp.Channel, nodeName string,
 	return
 }
 
+// MQSendRequest is the main function used for sending requests to our AMQP backend
 func MQSendRequest(channel *amqp.Channel, queue, operation, requestingUser, dbOwner, dbName, query string) (result []byte, err error) {
 	// Create a temporary AMQP queue for receiving the response
 	var q amqp.Queue

common/live.go

@@ -155,7 +155,7 @@ func RetrieveDatabaseFile(bucket, id string) (newDB string, err error) {
 	if _, err = os.Stat(newDB); os.IsNotExist(err) {
 		// * The database doesn't yet exist locally, so fetch it from Minio
 
-		// Check if a the database file is already being fetched from Minio by a different caller
+		// Check if the database file is already being fetched from Minio by a different caller
 		//  eg check if there is a "<filename>.new" file already in the disk cache
 		if _, err = os.Stat(newDB + ".new"); os.IsNotExist(err) {
 			// * The database isn't already being fetched, so we're ok to proceed
@@ -168,9 +168,7 @@ func RetrieveDatabaseFile(bucket, id string) (newDB string, err error) {
 			}
 
 			// Close the object handle when this function finishes
-			defer func() {
-				MinioHandleClose(userDB)
-			}()
+			defer MinioHandleClose(userDB)
 
 			// Create the needed directory path in the disk cache
 			err = os.MkdirAll(filepath.Join(Conf.DiskCache.Directory, bucket), 0750)

common/minio.go

@@ -1184,9 +1184,10 @@ func DisconnectPostgreSQL() {
 // If a non-0 discID value is passed, it will only return the details for that specific discussion/MR.  Otherwise it
 // will return a list of all discussions or MRs for a given database
 // Note - This returns a slice of DiscussionEntry, instead of a map.  We use a slice because it lets us use an ORDER
-//        BY clause in the SQL and preserve the returned order (maps don't preserve order).  If in future we no longer
-//        need to preserve the order, it might be useful to switch to using a map instead since they're often simpler
-//        to work with.
+//
+//	BY clause in the SQL and preserve the returned order (maps don't preserve order).  If in future we no longer
+//	need to preserve the order, it might be useful to switch to using a map instead since they're often simpler
+//	to work with.
 func Discussions(dbOwner, dbFolder, dbName string, discType DiscussionType, discID int) (list []DiscussionEntry, err error) {
 	dbQuery := `
 		WITH u AS (
@@ -1288,8 +1289,9 @@ func Discussions(dbOwner, dbFolder, dbName string, discType DiscussionType, disc
 // If a non-0 comID value is passed, it will only return the details for that specific comment in the discussion.
 // Otherwise it will return a list of all comments for a given discussion
 // Note - This returns a slice instead of a map.  We use a slice because it lets us use an ORDER BY clause in the SQL
-//        and preserve the returned order (maps don't preserve order).  If in future we no longer need to preserve the
-//        order, it might be useful to switch to using a map instead since they're often simpler to work with.
+//
+//	and preserve the returned order (maps don't preserve order).  If in future we no longer need to preserve the
+//	order, it might be useful to switch to using a map instead since they're often simpler to work with.
 func DiscussionComments(dbOwner, dbFolder, dbName string, discID, comID int) (list []DiscussionCommentEntry, err error) {
 	dbQuery := `
 		WITH u AS (

common/postgresql.go

@@ -430,6 +430,43 @@ func OpenSQLiteDatabaseDefensive(w http.ResponseWriter, r *http.Request, dbOwner
 		}
 	}
 
+	// Turn on cell size checking
+	err = sdb.Exec("PRAGMA cell_size_check=ON")
+	if err != nil {
+		log.Printf("Error when running 'PRAGMA cell_size_check=on' on the LIVE database: %s", err)
+		return
+	}
+
+	// Turn off mmap
+	var ok bool
+	var results []string
+	err = sdb.Select("PRAGMA mmap_size=0", func(s *sqlite.Stmt) error {
+		// Retrieve a row from the pragma result
+		var a string
+		if err = s.Scan(&a); err != nil {
+			// Error when reading the row, so return failure
+			ok = false
+			return err
+		}
+
+		// If the returned row was the value 0, then we regard the change as successfully applied.  Any other return
+		// value is a failure
+		switch a {
+		case "0":
+			ok = true
+		default:
+			ok = false
+			results = append(results, a)
+		}
+		return nil
+	})
+
+	// Check for a failure
+	if !ok || err != nil {
+		log.Printf("Error when disabling mmap on the LIVE database: %s\n", err)
+		return
+	}
+
 	// Set a SQLite authorizer which only allows SELECT statements to run
 	err = sdb.SetAuthorizer(AuthorizerSelect, "SELECT authorizer")
 	if err != nil {
@@ -459,6 +496,8 @@ func OpenSQLiteDatabaseDefensive(w http.ResponseWriter, r *http.Request, dbOwner
 // OpenSQLiteDatabaseLive is similar to OpenSQLiteDatabase(), but opens the a live SQLite database and implements
 // the recommended defensive precautions for potentially malicious user provided SQL
 // queries: https://www.sqlite.org/security.html
+// TODO: De-duplicate/refactor the common code in this function and OpenSQLiteDatabaseDefensive() above.  They're
+//       mostly the same
 func OpenSQLiteDatabaseLive(baseDir, dbOwner, dbName string) (sdb *sqlite.Conn, err error) {
 	dbPath := filepath.Join(baseDir, dbOwner, dbName, "live.sqlite")
 	if _, err = os.Stat(dbPath); err != nil {
@@ -561,9 +600,6 @@ func OpenSQLiteDatabaseLive(baseDir, dbOwner, dbName string) (sdb *sqlite.Conn,
 		}
 	}
 
-	// FIXME: There's new info in the SQLite "security" page, with some new things to potentially do: https://www.sqlite.org/security.html
-	//          * Disable triggers
-	// FIXME: Whatever improvements are made here, should probably also go into OpenSQLiteDatabaseDefensive()
 	// Turn on cell size checking
 	err = sdb.Exec("PRAGMA cell_size_check=ON")
 	if err != nil {
@@ -814,6 +850,93 @@ func ReadSQLiteDBCSV(sdb *sqlite.Conn, dbTable string) ([][]string, error) {
 	return resultSet, nil
 }
 
+// SQLiteBackupLive is used by our AMQP backend nodes to refresh a live SQLite database back into Minio
+func SQLiteBackupLive(baseDir, dbOwner, dbName string) (err error) {
+	dbPath := filepath.Join(baseDir, dbOwner, dbName, "live.sqlite")
+	if _, err = os.Stat(dbPath); err != nil {
+		return
+	}
+
+	// Open the database on the local node
+	// NOTE - OpenFullMutex seems like the right thing for ensuring multiple connections to a database file don't
+	// screw things up, but it wouldn't be a bad idea to keep it in mind if weirdness shows up
+	var sdb *sqlite.Conn
+	sdb, err = sqlite.Open(dbPath, sqlite.OpenReadWrite|sqlite.OpenFullMutex)
+	if err != nil {
+		log.Printf("Couldn't open LIVE database: %s", err)
+		return
+	}
+	if err = sdb.EnableExtendedResultCodes(true); err != nil {
+		log.Printf("Couldn't enable extended result codes for LIVE database query! Error: %v\n", err.Error())
+		return
+	}
+	defer sdb.Close()
+
+	// Generate unique temporary file name
+	var f *os.File
+	f, err = os.CreateTemp("", "sqliteBackup*.sqlite")
+	if err != nil {
+		return
+	}
+	tmpName := f.Name()
+	err = f.Close()
+	if err != nil {
+		return
+	}
+
+	// Generate the backup file
+	err = sdb.Exec(fmt.Sprintf("VACUUM INTO '%s'", tmpName))
+	if err != nil {
+		return
+	}
+
+	// Ensure the backup file was generated, and isn't 0 bytes (just in case)
+	var fileInfo os.FileInfo
+	fileInfo, err = os.Stat(tmpName)
+	if err != nil {
+		return
+	}
+	if fileInfo.Size() == 0 {
+		err = errors.New("Generating backup live SQLite database failed.  File size is 0")
+		return
+	}
+
+	// Sanity check the backup file
+	var sdb2 *sqlite.Conn
+	sdb2, err = sqlite.Open(tmpName, sqlite.OpenReadOnly)
+	if err != nil {
+		log.Printf("Couldn't open live database backup to sanity check it: %s", err)
+		return
+	}
+	if err = sdb2.EnableExtendedResultCodes(true); err != nil {
+		log.Printf("Couldn't enable extended result codes for live database backup integrity check! Error: %v", err.Error())
+		sdb2.Close()
+		return
+	}
+	// Pretty sure the '1' parameter below isn't needed.  The SQLite docs mention "O(N)" (etc.) which is just Big O
+	// notation, rather than trying to communicate a need for a number in the parameters
+	err = sdb2.IntegrityCheck("main", 1, true )
+	if err != nil {
+		sdb2.Close()
+		return
+	}
+	sdb2.Close()
+
+	// Copy the local backup file into Minio
+	var z *os.File
+	z, err = os.Open(tmpName)
+	err = LiveStoreDatabaseMinio(z, dbOwner, dbName, fileInfo.Size())
+	if err != nil {
+		z.Close()
+		return
+	}
+	z.Close()
+
+	// Delete the backup file from local disk
+	os.Remove(tmpName)
+	return
+}
+
 // SQLiteExecuteQueryLive is used by our AMQP backend infrastructure to execute a user provided SQLite statement
 func SQLiteExecuteQueryLive(baseDir, dbOwner, dbName, loggedInUser, query string) (rowsChanged int, err error) {
 	// Open the Live database on the local node

common/sqlite.go

@@ -55,8 +55,9 @@ const MaxLicenceSize = 1
 
 // MinioFolderChars is the number of leading characters of a files' sha256 used as the Minio folder name
 // eg: When set to 6, then "34f4255a737156147fbd0a44323a895d18ade79d4db521564d1b0dbb8764cbbc"
-//        -> Minio folder: "34f425"
-//        -> Minio filename: "5a737156147fbd0a44323a895d18ade79d4db521564d1b0dbb8764cbbc"
+//
+//	-> Minio folder: "34f425"
+//	-> Minio filename: "5a737156147fbd0a44323a895d18ade79d4db521564d1b0dbb8764cbbc"
 const MinioFolderChars = 6
 
 // QuerySource is used internally to help choose the output format from a SQL query

common/types.go

@@ -907,7 +907,7 @@ func DownloadDatabase(w http.ResponseWriter, r *http.Request, dbOwner, dbFolder,
 	w.Header().Set("Content-Disposition", fmt.Sprintf(`attachment; filename="%s"`, dbName))
 	w.Header().Set("Content-Length", fmt.Sprintf("%d", stat.Size))
 	w.Header().Set("Content-Type", "application/x-sqlite3")
-	bytesWritten, err = io.Copy(w, userDB)
+	_, err = io.Copy(w, userDB)
 	if err != nil {
 		log.Printf("Error returning DB file: %v\n", err)
 		return

common/util.go

@@ -779,18 +779,17 @@ func metadataGetHandler(w http.ResponseWriter, r *http.Request) {
 
 // postHandler receives uploaded files from DB4S. To simulate a DB4S upload, the following curl command can be used:
 //
-//   $ curl -kE ~/my.cert.pem -D headers.out -F [email protected] -F "branch=main" -F "commitmsg=stuff" \
-//       -F "sourceurl=https://example.org" -F "lastmodified=2017-01-02T03:04:05Z"  -F "licence=CC0"  -F "public=true" \
-//       https://db4s.dbhub.io:5550/someuser
+//	$ curl -kE ~/my.cert.pem -D headers.out -F [email protected] -F "branch=main" -F "commitmsg=stuff" \
+//	    -F "sourceurl=https://example.org" -F "lastmodified=2017-01-02T03:04:05Z"  -F "licence=CC0"  -F "public=true" \
+//	    https://db4s.dbhub.io:5550/someuser
 //
 // Subsequent uploads to the same database name will need to include an additional "commit" field, with the value of
 // the commit ID last known to DB4S.  An example curl command demonstrating this:
 //
-//   $ curl -kE ~/my.cert.pem -D headers.out -F [email protected] -F "branch=main" -F "commitmsg=stuff" \
-//       -F "sourceurl=https://example.org" -F "lastmodified=2017-01-02T03:04:05Z"  -F "licence=CC0"  -F "public=true" \
-//       -F "commit=51d494f2c5eb6734ddaa204eccb9597b426091c79c951924ac83c72038f22b55" \
-//       https://db4s.dbhub.io:5550/someuser
-//
+//	$ curl -kE ~/my.cert.pem -D headers.out -F [email protected] -F "branch=main" -F "commitmsg=stuff" \
+//	    -F "sourceurl=https://example.org" -F "lastmodified=2017-01-02T03:04:05Z"  -F "licence=CC0"  -F "public=true" \
+//	    -F "commit=51d494f2c5eb6734ddaa204eccb9597b426091c79c951924ac83c72038f22b55" \
+//	    https://db4s.dbhub.io:5550/someuser
 func postHandler(w http.ResponseWriter, r *http.Request, userAcc string) {
 	// Set the maximum accepted database size for uploading
 	oversizeAllowed := false
@@ -864,8 +863,7 @@ func postHandler(w http.ResponseWriter, r *http.Request, userAcc string) {
 
 // Returns a file requested by the client.  An example curl command to simulate the request is:
 //
-//   $ curl -OL -kE ~/my.cert.pem -D headers.out -G https://db4s.dbhub.io:5550/someuser/somedb.sqlite
-//
+//	$ curl -OL -kE ~/my.cert.pem -D headers.out -G https://db4s.dbhub.io:5550/someuser/somedb.sqlite
 func retrieveDatabase(w http.ResponseWriter, r *http.Request, pageName string, userAcc string, dbOwner string,
 	dbFolder string, dbName string, branchName string, commit string) (err error) {
 	pageName += ":retrieveDatabase()"
@@ -970,8 +968,7 @@ func rootHandler(w http.ResponseWriter, r *http.Request) {
 
 // Returns the list of databases available to the user.  To simulate, the following curl command can be used:
 //
-//   $ curl -kE ~/my.cert.pem -D headers.out -G https://db4s.dbhub.io:5550/someuser
-//
+//	$ curl -kE ~/my.cert.pem -D headers.out -G https://db4s.dbhub.io:5550/someuser
 func userDatabaseList(userAcc string, user string) (dbList []byte, err error) {
 
 	// Structure to hold the results, to apply JSON marshalling to

db4s/main.go

@@ -134,7 +134,7 @@ func main() {
 			err = json.Unmarshal(msg.Body, &req)
 			if err != nil {
 				log.Println(err)
-				err = com.MQErrorResponse(msg, ch, com.Conf.Live.Nodename, err.Error())
+				err = com.MQErrorResponse("NOT-YET-DETERMINED", msg, ch, com.Conf.Live.Nodename, err.Error())
 				if err != nil {
 					log.Printf("Error: occurred on '%s' the main live node switch{} while constructing an AMQP error message response: '%s'", com.Conf.Live.Nodename, err)
 				}
@@ -147,6 +147,23 @@ func main() {
 
 			// Handle each operation
 			switch req.Operation {
+			case "backup":
+				err = com.SQLiteBackupLive(com.Conf.Live.StorageDir, req.DBOwner, req.DBName)
+				if err != nil {
+					err = com.MQErrorResponse("BACKUP", msg, ch, com.Conf.Live.Nodename, err.Error())
+					if err != nil {
+						log.Printf("Error: occurred on '%s' in MQErrorResponse() while constructing an AMQP error message response: '%s'", com.Conf.Live.Nodename, err)
+					}
+					continue
+				}
+
+				// Return a success message to the caller
+				err = com.MQErrorResponse("BACKUP", msg, ch, com.Conf.Live.Nodename, "") // Use an empty error message to indicate success
+				if err != nil {
+					log.Printf("Error: occurred on '%s' in MQErrorResponse() while constructing the AMQP backup response: '%s'", com.Conf.Live.Nodename, err)
+				}
+				continue
+
 			case "columns":
 				var columns []sqlite.Column
 				columns, err = com.SQLiteGetColumnsLive(com.Conf.Live.StorageDir, req.DBOwner, req.DBName, req.Query) // We use the req.Query field to pass the table name