APRENDER. PRATICAR. EVOLUIR.O estado ativo pode exceder um ficheiro
A fixture cria uma aplicação SQLite com um lote e entradas de valores inteiros. O total inicial é 100, representado pela sequência 1. Ativa WAL, desativa o checkpoint automático apenas nesta área temporária e executa um checkpoint explícito da base inicial. Depois confirma uma entrada de 25 e atualiza o total para 125. O runner observa que o hash do ficheiro principal não mudou e que existe conteúdo no WAL. Ao copiar apenas o principal, obtém uma base legível que ainda contém a sequência 1. Esta experiência não depende de corrupção aleatória nem de falha física. Mostra que o estado confirmado de uma base ativa pode estar distribuído por componentes que um simples comando de cópia não interpreta. A existência do ficheiro e o sucesso da sua abertura são evidências insuficientes para concluir atualidade.
Copiar através do mecanismo suportado
A segunda cópia usa Connection.backup de Python, que disponibiliza a API de backup SQLite. A origem permanece aberta, mas o runner não introduz escritores concorrentes durante essa chamada. A cópia inclui a sequência 2 e o total 125. O destino é fechado e preparado como artefacto autónomo antes da sonda. Num ensaio separado, a ligação escritora altera o total para 999 sem confirmar a transação; outra ligação faz o backup e continua a obter 125. A alteração pendente é revertida no fim desse ensaio. Esta distinção ensina a observar estado confirmado sem transformar a experiência num benchmark de concorrência. A API permite procedimentos incrementais, mas o desempenho sob carga, erros de bloqueio e política de retry precisam de ensaios próprios. Regista conclusão, erros e critérios do artefacto em vez de aceitar apenas a sua presença.
Distinguir consistência de atualidade
Depois do backup, a origem confirma mais 75, passando para a sequência 3 e total 200. O artefacto anterior conserva sequência 2 e total 125. Os seus valores continuam coerentes entre si, mas falha quando a sonda exige sequência 3. Um ponto consistente pode ser demasiado antigo para o objetivo de recuperação. Na fixture, o requisito é uma sequência fornecida pelo ensaio, não calculada a partir da própria cópia. Num sistema real, acorda a referência, o seu produtor e a relação com o corte operacional. Sem timestamps e sem uma definição temporal, a diferença entre sequências não permite afirmar quantos minutos de dados se perderam. Do mesmo modo, a rapidez de uma consulta não mede o RTO completo. Comunica ponto disponível, ponto necessário e lacuna observada separadamente.
Transferir o raciocínio para operações
Num cenário fictício de fundos, o responsável de infraestrutura entrega o artefacto e o responsável da aplicação confirma se contém as operações necessárias. O procedimento deve indicar estado dos escritores, método suportado, localização, dependências e critérios de rejeição. Não elimines manualmente o WAL para reduzir o backup nem combines ficheiros ativos copiados em momentos diferentes como se isso provasse consistência. WAL tem também requisitos de coordenação: a documentação SQLite restringe a sua utilização a processos no mesmo host e não suporta o desenho habitual de vários hosts numa partilha de rede. Este comportamento pertence a SQLite; não é um procedimento Oracle, DB2 ou universal de storage. O código abaixo usa apenas ficheiros descartáveis em Darwin. Não formata volumes, altera mounts, executa comandos de dispositivos ou acede a uma base de produção.
#!/usr/bin/env python3
"""Original bounded SQLite backup/restore lab. Disposable local files only; no production database."""
import argparse,hashlib,json,platform,shutil,sqlite3,subprocess,sys,tempfile
from pathlib import Path
def probe(file,required):
db=sqlite3.connect(Path(file).resolve().as_uri()+'?mode=ro',uri=True)
try:
integrity=[x[0]for x in db.execute('PRAGMA integrity_check')]
foreign=list(map(list,db.execute('PRAGMA foreign_key_check')))
version=db.execute('PRAGMA user_version').fetchone()[0]
rows=list(map(list,db.execute('SELECT id,batch_id,amount FROM entries ORDER BY id')))
declared=db.execute('SELECT declared_total FROM batches WHERE id=1').fetchone()[0]
actual=sum(x[2]for x in rows if x[1]==1);latest=max((x[0]for x in rows),default=0)
conditions=dict(structural=integrity==['ok'],foreignKeys=not foreign,supportedVersion=version==1,totals=actual==declared,freshness=latest>=required)
return dict(ready=all(conditions.values()),conditions=conditions,integrity=integrity,foreignKeyFailures=foreign,schemaVersion=version,rows=rows,declaredTotal=declared,actualTotal=actual,latestSequence=latest,requiredSequence=required)
finally:db.close()
def main(output):
checks=[];observations={};calls=[]
def check(name,condition):assert condition,name;checks.append(name)
def digest(file):return hashlib.sha256(Path(file).read_bytes()).hexdigest()
def inspect(label,file,required):
result=subprocess.run([sys.executable,str(Path(__file__).resolve()),'--probe',str(file),'--required',str(required)],capture_output=True,text=True,timeout=15)
calls.append(dict(label=label,exit=result.returncode,stdout=result.stdout,stderr=result.stderr))
check(label+': probe exited successfully',result.returncode==0)
return json.loads(result.stdout)
def backup(source,destination):
target=sqlite3.connect(destination)
try:source.backup(target);target.execute('PRAGMA journal_mode=DELETE').fetchone()
finally:target.close()
with tempfile.TemporaryDirectory(prefix='dr-storage-recovery-')as temporary:
root=Path(temporary);live=root/'live.sqlite';db=sqlite3.connect(live,isolation_level=None)
try:
check('WAL mode enabled',db.execute('PRAGMA journal_mode=WAL').fetchone()[0]=='wal')
db.execute('PRAGMA wal_autocheckpoint=0');db.execute('PRAGMA foreign_keys=ON')
db.executescript('''BEGIN; PRAGMA user_version=1;
CREATE TABLE batches(id INTEGER PRIMARY KEY,declared_total INTEGER NOT NULL);
CREATE TABLE entries(id INTEGER PRIMARY KEY,batch_id INTEGER NOT NULL REFERENCES batches(id),amount INTEGER NOT NULL);
INSERT INTO batches VALUES(1,100);INSERT INTO entries VALUES(1,1,100);COMMIT;''')
checkpoint=db.execute('PRAGMA wal_checkpoint(TRUNCATE)').fetchone();check('baseline checkpoint completed',checkpoint[0]==0)
baselineHash=digest(live)
db.executescript('BEGIN;INSERT INTO entries VALUES(2,1,25);UPDATE batches SET declared_total=125 WHERE id=1;COMMIT;')
wal=Path(str(live)+'-wal');check('committed change remains in WAL',wal.exists()and wal.stat().st_size>0 and digest(live)==baselineHash)
raw=root/'main-only.sqlite';shutil.copyfile(live,raw);stale=inspect('main-only-copy',raw,2)
check('main-only copy structurally valid',stale['conditions']['structural'])
check('main-only copy misses committed sequence',stale['latestSequence']==1 and not stale['ready']and not stale['conditions']['freshness'])
observations['mainOnly']=dict(result=stale,mainBytesUnchanged=True,walBytes=wal.stat().st_size,sourceConnectionOpen=True)
saved=root/'online.sqlite';backup(db,saved);snapshot=inspect('online-backup',saved,2)
check('online backup includes committed WAL content',snapshot['ready']and snapshot['latestSequence']==2 and snapshot['actualTotal']==125)
observations['onlineBackup']=snapshot
db.execute('BEGIN IMMEDIATE');db.execute('UPDATE batches SET declared_total=999 WHERE id=1')
reader=sqlite3.connect(live)
try:uncommitted=root/'uncommitted-excluded.sqlite';backup(reader,uncommitted)
finally:reader.close();db.execute('ROLLBACK')
excluded=inspect('uncommitted-excluded',uncommitted,2)
check('separate backup excludes uncommitted update',excluded['ready']and excluded['declaredTotal']==125)
observations['uncommittedExcluded']=excluded
db.executescript('BEGIN;INSERT INTO entries VALUES(3,1,75);UPDATE batches SET declared_total=200 WHERE id=1;COMMIT;')
older=inspect('snapshot-after-new-commit',saved,3)
check('snapshot remains at earlier generation',older['latestSequence']==2 and older['conditions']['totals']and not older['conditions']['freshness'])
observations['laterCommit']=dict(snapshot=older,liveLatest=db.execute('SELECT max(id) FROM entries').fetchone()[0],liveTotal=db.execute('SELECT declared_total FROM batches').fetchone()[0])
db.close();live.unlink();check('synthetic original removed before restore',not live.exists())
restored=root/'restored.sqlite';shutil.copyfile(saved,restored);before=digest(restored);accepted=inspect('restored-application',restored,2);after=digest(restored)
check('restored application passes explicit contract',accepted['ready']and accepted['actualTotal']==125)
check('read-only restore probe preserves artifact',before==after)
observations['restored']=dict(result=accepted,hashBefore=before,hashAfter=after,originalLiveNotRequired=True)
broken=root/'business-defect.sqlite';shutil.copyfile(saved,broken);edit=sqlite3.connect(broken)
try:edit.execute('UPDATE batches SET declared_total=999 WHERE id=1');edit.commit()
finally:edit.close()
bad=inspect('business-defect',broken,2)
check('integrity check misses business-total defect',bad['conditions']['structural']and bad['conditions']['foreignKeys']and not bad['conditions']['totals']and not bad['ready'])
observations['businessDefect']=bad
orphan=root/'foreign-key-defect.sqlite';shutil.copyfile(saved,orphan);edit=sqlite3.connect(orphan)
try:edit.execute('PRAGMA foreign_keys=OFF');edit.execute('INSERT INTO entries VALUES(3,999,10)');edit.commit()
finally:edit.close()
fk=inspect('foreign-key-defect',orphan,2)
check('foreign_key_check catches defect beyond integrity_check',fk['conditions']['structural']and not fk['conditions']['foreignKeys']and not fk['ready'])
observations['foreignKeyDefect']=fk
incompatible=root/'unsupported-version.sqlite';shutil.copyfile(saved,incompatible);edit=sqlite3.connect(incompatible)
try:edit.execute('PRAGMA user_version=2');edit.commit()
finally:edit.close()
unsupported=inspect('unsupported-version',incompatible,2)
check('application rejects unsupported schema marker',unsupported['conditions']['structural']and unsupported['conditions']['totals']and not unsupported['conditions']['supportedVersion']and not unsupported['ready'])
observations['unsupportedVersion']=unsupported
missing=root/'missing.sqlite'
try:probe(missing,2)
except sqlite3.OperationalError:rejected=True
else:rejected=False
check('read-only missing path fails without creation',rejected and not missing.exists())
observations['missingPath']=dict(openRejected=rejected,fileCreated=missing.exists())
finally:db.close()
check('temporary database artifacts removed',not root.exists())
report=dict(python=platform.python_version(),sqlite=sqlite3.sqlite_version,system=platform.system(),release=platform.release(),machine=platform.machine(),runnerSha256=hashlib.sha256(Path(__file__).read_bytes()).hexdigest(),checks=checks,observations=observations,probeProcesses=calls,scope='Original local SQLite fixture on the recorded host. Actual WAL copy, backup API, isolated restore probes and deliberately introduced defects. Sequence thresholds are synthetic acceptance criteria, not measured RPO/RTO. No external service, Oracle/DB2 behavior, Linux execution, network filesystem, power failure, device change, load benchmark, encryption-key recovery or production data.')
if output:Path(output).write_text(json.dumps(report,indent=2)+'\n')
print(json.dumps(dict(groups=len(observations),checks=len(checks),probeProcesses=len(calls),sqlite=sqlite3.sqlite_version,output=output)))
if __name__=='__main__':
p=argparse.ArgumentParser();p.add_argument('--output');p.add_argument('--probe');p.add_argument('--required',type=int,default=2);a=p.parse_args()
if a.probe:print(json.dumps(probe(a.probe,a.required)))
else:main(a.output)
Tejo copia o ficheiro principal durante atividade. A cópia abre, mas falta uma operação já confirmada; o backup suportado recupera a sequência exigida.
Armadilhas comuns
Tratar um ficheiro copiado como toda a base ativa, apagar o WAL, aceitar por tamanho ou confundir uma sequência com uma medição de minutos de perda.
Tópicos relacionados: Snapshots e recuperação · Transações e dependências
Um artefacto pode ser estruturalmente válido e estar desatualizado. A cópia suportada e a validação do ponto recuperável respondem a perguntas diferentes.
Referência: SQLite online backup API · DR Storage 2026-09; selected Linux and AWS storage behavior